如何基于MassTransit设计自定义消息系统并减少队列与交换数量？

1. 按业务域/流程分组，共享接收端点 #

如果多个操作属于同一业务流程或领域，可以将它们绑定到同一个ReceiveEndpoint，这样只会生成一组Exchange和Queue。配置示例：

cfg.ReceiveEndpoint("order-processing-queue", e =>
{
    // 把同业务域的消费者都挂载到这个端点
    e.Consumer<CreateOrderConsumer>();
    e.Consumer<UpdateOrderConsumer>();
    e.Consumer<CancelOrderConsumer>();

    // 关闭自动为每个消息类型创建交换，手动绑定统一的业务交换
    e.BindMessageExchanges = false;
    e.Bind("order-events-exchange", x =>
    {
        x.RoutingKey = "order.*"; // 用通配符匹配所有订单相关消息
        x.ExchangeType = ExchangeType.Topic;
    });
});

2. 利用主题交换（Topic Exchange）做集中路由 #

借助RabbitMQ的主题交换特性，将多个消息类型路由到同一队列，大幅减少交换数量：

• 为不同消息类型设置规范的路由键（如order.created、payment.completed）
• 声明全局主题交换，接收端点通过通配符绑定需要处理的路由键

配置示例：

// 发布端指定消息的路由键
cfg.Message<CreateOrder>(x => x.SetEntityName("order.created"));
cfg.Message<PaymentCompleted>(x => x.SetEntityName("payment.completed"));

// 接收端绑定全局主题交换
cfg.ReceiveEndpoint("payment-handler-queue", e =>
{
    e.Consumer<ProcessPaymentConsumer>();
    e.Consumer<RefundPaymentConsumer>();

    e.Bind("business-events-topic", x =>
    {
        x.RoutingKey = "payment.*";
        x.ExchangeType = ExchangeType.Topic;
    });
});

3. 禁用自动创建消息交换 #

默认MassTransit会为每个消息类型单独创建交换并绑定到端点，关闭该特性后可手动控制交换数量：

cfg.ReceiveEndpoint("shared-business-queue", e =>
{
    e.BindMessageExchanges = false; // 关闭自动绑定逻辑
    // 手动绑定需要的集中交换
    e.Bind("core-events-exchange", x =>
    {
        x.RoutingKey = "#"; // 匹配所有路由键（按需使用，避免滥用）
    });
    // 挂载多个消费者
    e.Consumer<EventAConsumer>();
    e.Consumer<EventBConsumer>();
});

4. 批量消费者/消息分发逻辑 #

如果多个操作逻辑相似，可通过单个消费者根据消息子类型分发处理，进一步简化端点配置：

cfg.ReceiveEndpoint("unified-processing-queue", e =>
{
    e.Consumer<UnifiedEventConsumer>(x =>
    {
        x.Message<BaseBusinessEvent>(context =>
        {
            var message = context.Message;
            if (message is CreateOrder order)
                HandleCreateOrder(order);
            else if (message is UpdateOrder update)
                HandleUpdateOrder(update);
        });
    });
});

注意事项 #

• 共享队列需根据业务吞吐量调整并发数（如e.PrefetchCount = 15;），避免处理瓶颈
• 路由键设计要遵循规范，避免模糊匹配导致消息错误路由
• 生产环境提前制定交换/队列命名规则（如{domain}-{process}-exchange），便于运维管理

内容的提问来源于stack exchange，提问作者Majid Gholipour