MassTransit单个消费者能否绑定多个RabbitMQ端点？配置遇问题

我尝试将单个消费者绑定到多个端点，这是否可行？

我的消费者代码如下：

class OrderPlacedConsumer : IConsumer<OrderPlaced>
{
    public Task Consume(ConsumeContext<OrderPlaced> context)
    {
        //do stuff
    }
}

以下是我注册消费者和总线的方式：

services.AddMassTransit(x =>
{
    x.AddConsumer<OrderPlacedConsumer>();

    x.UsingRabbitMq((ctx, cfg) =>
    {
        //Read RabbitMq setting from config 
        string connectionString = config.ConnectionStrings.RabbitMQConnString;
        RabbitMQSetting rabbitMQSetting = GetRabbitMQSetting(connectionString);
        cfg.Host(rabbitMQSetting.Host, (ushort)rabbitMQSetting.Port, "/", h =>
        {
            h.PublisherConfirmation = false;
            h.Username(rabbitMQSetting.UserName);
            h.Password(rabbitMQSetting.Password);
            h.Heartbeat(TimeSpan.FromSeconds(rabbitMQSetting.Heartbeat));
        });

        cfg.ReceiveEndpoint("endpoint1", e =>
        {
            e.ConfigureConsumer(ctx, OrderPlacedConsumer);

            #region Exchange
            e.ConfigureConsumeTopology = false;
            e.Bind("OrderExchange", x =>
            {
                x.ExchangeType = ExchangeType.Direct;
                x.RoutingKey = "endpoint1";
            });
            #endregion
        });

        cfg.ReceiveEndpoint("endpoint2", e =>
        {
            e.ConfigureConsumer(ctx, OrderPlacedConsumer);

            #region Exchange
            e.ConfigureConsumeTopology = false;
            e.Bind("OrderExchange", x =>
            {
                x.ExchangeType = ExchangeType.Direct;
                x.RoutingKey = "endpoint2";
            });
            #endregion
        });
    });
});

但目前OrderPlacedConsumer仅绑定到了endpoint1，请问能否将其同时绑定到endpoint2？

解决方案 #

完全可以实现单个消费者绑定多个端点，问题出在配置代码的ConfigureConsumer调用方式上——你直接传递了OrderPlacedConsumer类型，没有正确从上下文获取消费者的配置信息，导致只有第一个端点生效。

修改步骤 #

把两个ReceiveEndpoint中的e.ConfigureConsumer(ctx, OrderPlacedConsumer);替换为：

e.ConfigureConsumer<OrderPlacedConsumer>(ctx);

修改后的完整配置代码：

services.AddMassTransit(x =>
{
    x.AddConsumer<OrderPlacedConsumer>();

    x.UsingRabbitMq((ctx, cfg) =>
    {
        //Read RabbitMq setting from config 
        string connectionString = config.ConnectionStrings.RabbitMQConnString;
        RabbitMQSetting rabbitMQSetting = GetRabbitMQSetting(connectionString);
        cfg.Host(rabbitMQSetting.Host, (ushort)rabbitMQSetting.Port, "/", h =>
        {
            h.PublisherConfirmation = false;
            h.Username(rabbitMQSetting.UserName);
            h.Password(rabbitMQSetting.Password);
            h.Heartbeat(TimeSpan.FromSeconds(rabbitMQSetting.Heartbeat));
        });

        cfg.ReceiveEndpoint("endpoint1", e =>
        {
            e.ConfigureConsumer<OrderPlacedConsumer>(ctx);

            e.ConfigureConsumeTopology = false;
            e.Bind("OrderExchange", x =>
            {
                x.ExchangeType = ExchangeType.Direct;
                x.RoutingKey = "endpoint1";
            });
        });

        cfg.ReceiveEndpoint("endpoint2", e =>
        {
            e.ConfigureConsumer<OrderPlacedConsumer>(ctx);

            e.ConfigureConsumeTopology = false;
            e.Bind("OrderExchange", x =>
            {
                x.ExchangeType = ExchangeType.Direct;
                x.RoutingKey = "endpoint2";
            });
        });
    });
});

原理说明 #

MassTransit注册消费者时会生成对应的ConsumerDefinition配置，ConfigureConsumer<T>方法会从上下文获取这个定义，将消费者正确关联到接收端点。你之前的写法没有传递有效的配置引用，所以只有第一个端点完成了绑定。

内容的提问来源于Stack Exchange，提问作者Nikhil Bandivan