您如何使用 MassTransit.Multibus 配置 IRequestClient<T> 和 IConsumer<T>？

如何解决您如何使用 MassTransit.Multibus 配置 IRequestClient<T> 和 IConsumer<T>？

我们一直将 MassTransit 与我们服务内部的单个 RabbitMq 传输一起使用。我们有一个新的 RabbitMq 服务器，它是公共的，我们也想连接到某些事件，所以我们自然要使用 Multibus 功能。 连接成功，消息似乎发布得很好，但我们旧的 RequestClient 消费者似乎不再在原始总线上工作，我不知道为什么。抛出的错误是 MassTransit.RequestTimeoutException: Timeout waiting for response。 Multibus IBuse 应该自己启动，对吗？

以下是 Startup.cs ConfigureServices 中的样子（ICorrespondenceInternalBus 和 ICorrespondenceExternalBus 都继承自 IBus）：

...
//First bus
services.AddMassTransit<ICorrespondenceInternalBus>(c =>
{
    c.AddConsumersFromNamespaceContaining(GetType());
    ConfigureAdditionalMassTransitServices(c);

    c.UsingRabbitMq((context,cfg) =>
    {
        cfg.Host(new Uri($"rabbitmq://{rabbitMqServerName}:/"),h =>
            {
                h.Username("guest");
                h.Password("guest");
            });

        cfg.ConfigureEndpoints(context,new CorrespondenceSystemEndpointNameFormatter());
        cfg.UseMessageRetry(retryConfig => retryConfig.Interval(5,TimeSpan.FromMilliseconds(250)));
        cfg.UseHealthCheck(context);
    });
});
services.AddMassTransitHostedService();
...
//second bus
services.AddMassTransit<ICorrespondenceExternalBus>(c =>
{
    c.UsingRabbitMq((context,cfg) =>
    {
        cfg.Host(rabbitMqServerName,port,virtualHost,h =>
        {
            h.Username(username);
            h.Password(password);
            if (useSsl)
                h.UseSsl(s => s.Protocol = SslProtocols.Tls12);
        });
        cfg.MessageTopology.SetEntityNameFormatter(new CorrespondenceSystemExternalEntityNameFormatter());
        cfg.UseHealthCheck(context);
    });
});

在上面，rabbitmq 中的两个总线注册和 Exchange 似乎都接收已发布的消息。不工作的部分正在使用来自 RequestClients 的消息。 以下是 RequestClient 的注册方式：

protected override void ConfigureAdditionalMassTransitServices(
IServiceCollectionConfigurator<ICorrespondenceInternalBus> configurator)
{
    configurator.AddRequestClient<ICheckForDuplicateQuery>();
}

正在运行的 RequestHandler：

public class Handler : IRequestHandler<Command,Dto>
{
    private readonly IRequestClient<ICheckForDuplicateQuery> _duplicateCheckClient;

    public Handler(IRequestClient<ICheckForDuplicateQuery> duplicateCheckClient)
    {
        _duplicateCheckClient = duplicateCheckClient;
    }

    public async Task<Dto> Handle(Command request,CancellationToken cancellationToken)
    {
        var duplicateQuery = new Query();
        var duplicateCheckResult = await _duplicateCheckClient.GetResponse<ICheckForDuplicateQueryResult>(duplicateQuery,cancellationToken,TimeSpan.FromSeconds(10));

        if (duplicateCheckResult.Message.IsDuplicate)
            return new DuplicateDto(duplicateCheckResult.Message.CorrelationIds.First());

        ...
    }
}

最后是消费者：

public class CheckForDuplicateQueryHandler : IConsumer<ICheckForDuplicateQuery>
{    
    ...
    public async Task Consume(ConsumeContext<ICheckForDuplicateQuery> context)
    {
        if (context is null)
            throw new ArgumentNullException(nameof(context));
        ...
        await context.RespondAsync(new Result()).ConfigureAwait(false);
    }

    private class Result : ICheckForDuplicateQueryResult
    {
        ...
    }
}

消费者永不进入，请求客户端超时。 为了比较，这里是在我们尝试 Multibus 之前，当 RequestClients 工作正常时的样子（消费者和请求客户端逻辑完全相同，只有 Startup.cs 不同： 以前的（单总线）Startup.cs：

services.AddMassTransit(c =>
{
    c.AddConsumersFromNamespaceContaining(GetType());
    ConfigureAdditionalMassTransitServices(c);

    c.AddBus(provider => Bus.Factory.CreateUsingRabbitMq(sbc =>
    {
        sbc.Host(new Uri($"rabbitmq://{rabbitMqServerName}:/"),h =>
            {
                h.Username("guest");
                h.Password("guest");
            });

        sbc.ConfigureEndpoints(provider,new CorrespondenceSystemEndpointNameFormatter());
        sbc.UseMessageRetry(cfg => cfg.Interval(5,TimeSpan.FromMilliseconds(250)));
    }));
});
HealthChecksBuilder.AddRabbitMqHealthcheck(rabbitMqServerName);
...
public virtual void Configure(IApplicationBuilder app,IHostEnvironment env,IHostApplicationLifetime appLifetime,IBusControl bus)
{
    ...
    appLifetime.ApplicationStarted.Register(bus.Start);
    appLifetime.ApplicationStopping.Register(bus.Stop);
}
...
//registering the RequestClients previously:
protected override void ConfigureAdditionalMassTransitServices(IServiceCollectionBusConfigurator configurator)
{
    configurator.AddRequestClient<ICheckForDuplicateQuery>();
}

在此先感谢您的帮助！如果您需要查看更多代码片段，我很乐意提供它们，我试图保持简洁，仅包含更改中需要/影响的内容。

解决方法

我已经确认请求客户端应该使用正确的总线实例，具体取决于它的配置位置 in this unit test commit。

所以，我不确定为什么您没有看到相同的行为。