如何使用MassTransit建立与两个不同RabbitMQ服务器的连接？

MassTransit 支持通过多总线实例的方式对接多个不同的RabbitMQ服务器，你需要为每个独立的RabbitMQ集群定义单独的总线标记，分别注册配置即可，该方案兼容MassTransit 8及以上主流版本。

步骤1：定义独立的总线标记接口 #

为第二个RabbitMQ服务创建一个空的总线标记接口，用来区分不同的总线实例，可根据实际业务场景命名，避免用数字序号增加后期维护成本：

// 示例：对接订单服务RabbitMQ集群的总线标记
public interface IOrderRabbitMqBus : IBus
{
}

如果后续还要对接更多RabbitMQ服务，继续新增对应的标记接口即可。

步骤2：注册第二个总线实例 #

原有第一个RabbitMQ服务的注册逻辑无需修改，直接新增第二个总线的注册配置即可，各自独立配置连接参数、消费者、接收端点：

// 原有第一个RabbitMQ服务的注册保持不变
services.AddMassTransit(x =>
{ 
    x.AddConsumer<OptionExecutedConsumer>();
    
    x.UsingRabbitMq((context, cfg) =>
    {
        cfg.Host(rabbitMQConfig.Connection.Servers[0],
        rabbitMQConfig.Connection.Port,
        rabbitMQConfig.Connection.VirtualHost,
        h =>
        {
            h.Username(rabbitMQConfig.Connection.Username);
            h.Password(rabbitMQConfig.Connection.Password);
            h.UseCluster(c =>
            {
                rabbitMQConfig.Connection.Servers.ToList()
                    .ForEach(server => c.Node(server));
            });
        });
        cfg.ReceiveEndpoint(rabbitMQConfig.Consumer.OptionExerciseQueue, e =>
        {
            e.UseRetry(r =>
                r.Incremental(rabbitMQConfig.Consumer.RetryLimit,
                TimeSpan.FromSeconds(rabbitMQConfig.Consumer.InitialInterval),
                TimeSpan.FromSeconds(rabbitMQConfig.Consumer.IntervalIncrement)));
    
            e.Bind(rabbitMQConfig.Consumer.OptionExerciseExchange, x =>
            {
                e.ConcurrentMessageLimit = 1;
                e.DiscardFaultedMessages();
                e.ClearMessageDeserializers();
                e.UseRawJsonSerializer();
                e.ConfigureConsumer<OptionExecutedConsumer>(context);
            });
        });      
    });
});

// 新增第二个RabbitMQ服务的总线注册
services.AddMassTransit<IOrderRabbitMqBus>(x =>
{
    // 第二个RabbitMQ服务对应的消费者在这里单独注册
    x.AddConsumer<OrderCreatedConsumer>();
    
    x.UsingRabbitMq((context, cfg) =>
    {
        // 填入第二台RabbitMQ服务器的独立连接配置
        cfg.Host(orderRabbitMQConfig.Connection.Servers[0],
        orderRabbitMQConfig.Connection.Port,
        orderRabbitMQConfig.Connection.VirtualHost,
        h =>
        {
            h.Username(orderRabbitMQConfig.Connection.Username);
            h.Password(orderRabbitMQConfig.Connection.Password);
            // 第二台如果是单节点服务可删除集群配置
            h.UseCluster(c =>
            {
                orderRabbitMQConfig.Connection.Servers.ToList()
                    .ForEach(server => c.Node(server));
            });
        });
        
        // 配置第二台服务器对应的队列、Exchange绑定规则
        cfg.ReceiveEndpoint(orderRabbitMQConfig.Consumer.OrderCreatedQueue, e =>
        {
            e.UseRetry(r =>
                r.Incremental(orderRabbitMQConfig.Consumer.RetryLimit,
                TimeSpan.FromSeconds(orderRabbitMQConfig.Consumer.InitialInterval),
                TimeSpan.FromSeconds(orderRabbitMQConfig.Consumer.IntervalIncrement)));
    
            e.Bind(orderRabbitMQConfig.Consumer.OrderCreatedExchange, x =>
            {
                e.ConcurrentMessageLimit = 2;
                e.DiscardFaultedMessages();
                e.ClearMessageDeserializers();
                e.UseRawJsonSerializer();
                e.ConfigureConsumer<OrderCreatedConsumer>(context);
            });
        });      
    });
});

// 必须单独注册每个总线的宿主服务，负责总线的启动、停止
services.AddHostedService<MassTransitHostedService>();
services.AddHostedService<MassTransitHostedService<IOrderRabbitMqBus>>();

步骤3：不同总线的调用方式 #

如果需要向不同的RabbitMQ服务发送消息，注入对应的总线实例即可：

public class MessageService(IBus defaultBus, IOrderRabbitMqBus orderBus)
{
    // 向第一台RabbitMQ发消息
    public async Task SendToDefaultServer(OptionExecutedEvent message)
    {
        await defaultBus.Publish(message);
    }

    // 向第二台RabbitMQ发消息
    public async Task SendToOrderServer(OrderCreatedEvent message)
    {
        await orderBus.Publish(message);
    }
}

• 总线标记接口按业务域命名，不要用数字序号，方便后期快速识别每个总线对接的业务系统
• 不同总线的消息类、消费者不要跨总线复用，避免出现序列化异常、路由错误等问题
• 每个总线的重试策略、序列化规则、限流配置独立设置，不要复用同一套配置逻辑
• 生产环境为每个总线配置独立的健康检查规则，方便单独监控不同RabbitMQ服务的连接状态

内容的提问来源于stack exchange，提问作者Thiago Adriano