masstransit和多个总线实例

Question

根据MT文档，每个总线实例应具有不同的队列名称。

我是否正确地认为以下是正确的配置？

什么将WebApi和后端队列连接在一起？

考虑更高级的方案

[当我设计后端管道以同时使用消息处理时，我可以稍后对其进行切片，并使其非常容易地用于有线传输中。问题是，我可以以某种方式配置MT，以便Azure配置的总线将消息中继到配置了本地传输的总线吗？

Answer 1

我将在MassTransit v3的上下文中回答此问题，该版本同时支持Azure Service Bus和RabbitMQ。如果您要使用Azure，我强烈建议您在v2的传输版本上使用v3。

首先，关于请求。它们应该发送，而不是发布。通常，它们本质上是命令，而不是事件。我看到人们发布请求的唯一原因是他们没有服务的端点地址。因此了解端点会有很大帮助。

第二，在您的示例中，每个WebAPI实例都应该有自己的队列来接收响应，因为它们被发送回请求者。使用MT3，每个IBus实例都有一个唯一的自动删除队列，该队列为此目的而设置，以处理来自请求的响应。

在GitHub上的MassTransit存储库中有一个示例Sample-RequestResponse，显示了如何使用RabbitMQ进行设置。与Azure Service Bus差不多。

将所有设备组合在一起的“结构”是虚拟主机（在RabbitMQ中）或名称空间（在ASB中）。主题和队列之间的联系决定了它们如何共同构成逻辑bus。

Answer 2

如果有人尝试使用.NET Core + DI来注册multiple总线：

• 请勿使用AddBus中的内部版本
  • 无论您做什么，都不会注册多于一辆的公共汽车
  • 这是因为它在内部调用TryAddSingleton调用
  • [TryAddSingleton仅在尚未为接口注册实例的情况下，才将新实例添加到DI容器中
  • 注意：
    • 无异常或错误

简而言之是我们的解决方案

由于各种必要的接口不是通用的，我们：

• 围绕内置接口创建的通用包装器
• 创建了一个接口（使用通用参数），该接口唯一地标识每个RegisteredBus
• 当创建新的Wrapper实例时，我们将内置接口的实例传递到其构造函数中
• 包装器然后将内部内部接口的实例保存在公共属性TryAddSingleton中
• 而不是例如注入Instance，我们现在注入IBus
  • 然后我们使用包装器的IBus<MyRegisteredBus>属性来访问内置接口实例，并将其存储以供以后使用（此后包装器不再起作用）

代码

通用Instance调用（否则与内置调用具有100％相同的签名）：

AddBus

我们为实现此目的而创建的各种接口/类：

        public static void AddBus<TBusType>(this IServiceCollection services, Func<IServiceProvider, IBusControl> busFactory)
            where TBusType : class, IBusType
        {
            IBusControl<TBusType> BusFactory(IServiceProvider serviceProvider)
            {
                return new BusControl<TBusType>(busFactory(serviceProvider));
            }

            services.AddSingleton<IBusControl<TBusType>>(BusFactory);
            services.AddSingleton<IBus<TBusType>>(provider => new Bus<TBusType>(provider.GetRequiredService<IBusControl<TBusType>>().Instance));
        }

如何注册通用

    // the only purpose of the interfaces derived from `IBusType` is to uniquely idnetify a registered Bus
    public interface IBusType { }
    public interface IHosted : IBusType { }
    public interface ILocal : IBusType { }

    public interface IBusTypeWrapper<TBusType, TInterface>
        where TBusType : IBusType
    {
        public TInterface Instance { get; }
    }

    public class BusTypeWrapper<TBusType, TInterface> : IBusTypeWrapper<TBusType, TInterface>
        where TBusType : IBusType
    {
        public TInterface Instance { get; }

        public BusTypeWrapper(TInterface instance)
        {
            Instance = instance;
        }
    }

    public interface IBusControl<T> : IBusTypeWrapper<T, IBusControl> where T : IBusType { }
    public class BusControl<T> : BusTypeWrapper<T, IBusControl>, IBusControl<T> where T : IBusType
    {
        public BusControl(IBusControl instance) : base(instance) { }
    }

    public interface IBus<T> : IBusTypeWrapper<T, IBus> where T : IBusType { }
    public class Bus<T> : BusTypeWrapper<T, IBus>, IBus<T> where T : IBusType
    {
        public Bus(IBus instance) : base(instance) { }
    }

    public interface ISendEndpointProvider<T> : IBusTypeWrapper<T, ISendEndpointProvider> where T : IBusType { }
    public class SendEndpointProvider<T> : BusTypeWrapper<T, ISendEndpointProvider>, ISendEndpointProvider<T> where T : IBusType
    {
        public SendEndpointProvider(ISendEndpointProvider instance) : base(instance) { }
    }

：

ISendEndpointProvider