公共交通和多个总线实例

Question

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

我可以假设以下是正确的配置吗？

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

考虑更高级的场景

当我设计我的后端管道以使用消息处理时，我可以稍后对其进行切片并让它很容易地通过有线传输使用。问题是，我能否以某种方式配置 MT，以便 Azure 配置的总线将消息中继到配置有本地传输的总线？

Answer 1

我将在 MassTransit v3 的上下文中回答这个问题，它同时支持 Azure 服务总线和 RabbitMQ。如果你打算使用 Azure，我强烈推荐 v3 而不是 v2 的传输。

首先，关于请求。它们应该发送，而不是发布。它们通常是天生的命令，而不是事件。我看到人们发布请求的唯一原因是他们没有服务的端点地址。所以知道端点有很大帮助。

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

在 GitHub 上的 MassTransit 存储库中有一个示例 Sample-RequestResponse，它显示了如何使用 RabbitMQ 进行设置。 Azure 服务总线也差不多。

将所有这些结合在一起的“结构”是虚拟主机（在 RabbitMQ 中）或命名空间（在 ASB 中）。主题和队列之间的联系决定了它们如何一起工作以形成一个逻辑总线。

Answer 2

如果有人试图让我们 .NET Core + DI 注册 多个 总线：

• 请勿在AddBus 调用中使用构建
  • 无论您做什么，都不会注册超过一辆巴士到期
  • 这是因为它在内部调用TryAddSingleton调用
  • TryAddSingleton 仅在没有为接口注册实例的情况下才向 DI 容器添加新实例
  • 注意：
    • 不抛出异常或错误

我们使用的解决方案

由于所需的各种接口不是通用的：

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

我们希望不必使用带有奇怪Instance 属性的某种包装器，但如果内置接口变得通用或使用类似DynamicProxies 的东西，我们无法想出更优雅的解决方案。

非常欢迎提出想法/反馈。

代码

通用 AddBus 调用（否则 100% 与内置调用相同的签名）：

        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：

services.AddSingleton<ISendEndpointProvider<ILocal>>(provider => new SendEndpointProvider<ILocal>(provider.GetRequiredService<IBusControl<ILocal>>().Instance));

更新

对于每种总线类型的 IHosted 服务： - 创建一个通用的HostedService<BusType> 服务 - 在构造函数中注入IBusControl<BusType> - 并使用注入的实例开始停止特定的总线

然后为每种总线类型注册一个IHostedService。

services.AddSingleton<IHostedService, HostedService<ILocal>>(); services.AddSingleton<IHostedService, HostedService<IHosted>>();`