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

Question

我们一直在将 MassTransit 与我们服务内部的单个 RabbitMq 传输一起使用。我们有一个新的 RabbitMq 服务器，它是公共的，我们也想为某些事件连接它，所以自然我们想使用 Multibus 功能。 连接成功，消息似乎可以正常发布，但我们的旧 RequestClient 消费者似乎不再在原始总线上工作，我不知道为什么。抛出的错误是MassTransit.RequestTimeoutException: Timeout waiting for response。多总线 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 的消息。 以下是 RequestClients 的注册方式：

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>();
}

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

Answer 1

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

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