在C#中使用MassTransit和RabbitMQ实现多环境队列消息发布

针对你的场景，要在发布端指定目标队列，有几种直接可行的实现方式，结合你的现有代码整理如下：

方法1：通过EndpointConvention绑定消息到指定队列 #

可以在MassTransit配置阶段，为消息类型绑定固定的目标队列，这样每次发布该消息时都会自动发送到指定队列，适合按环境统一配置的场景。

修改你的Program.cs配置：

builder.Services.AddMassTransit(c =>
{
    c.UsingRabbitMq((context, configurator) =>
    {
        configurator.Host(builder.Configuration["EventBusSettings:HostAddress"]);

        // 获取当前环境的目标队列名
        var targetQueue = builder.Configuration["EventBusSettings:Queue"] ?? string.Empty;
        
        // 为消息类型绑定对应队列的Exchange地址
        EndpointConvention.Map<PublishAssetRequest>(new Uri($"exchange:{targetQueue}"));
        EndpointConvention.Map<PublishAssetMetadataRequest>(new Uri($"exchange:{targetQueue}"));
    });
});

配置完成后，你原有的_publishEndpoint.Publish(publishAssetRequest)代码无需修改，会自动发送到指定队列。

方法2：发布时动态指定目标队列地址 #

如果需要在发布环节动态指定队列，可以使用Publish的重载方法，直接传入目标队列的地址。

修改你的PublishController代码：

PublishAssetRequest publishAssetRequest = new(){ 
    AssetId = asset.Id, 
    ApplicationId = createAssetRequest.ApplicationId 
};

// 读取配置中的目标队列名，生成对应地址
var targetQueueAddress = new Uri($"exchange:{builder.Configuration["EventBusSettings:Queue"]}");

// 发布时指定目标地址
await _publishEndpoint.Publish(publishAssetRequest, context =>
{
    context.SendEndpointAddress = targetQueueAddress;
});

也可以通过获取SendEndpoint的方式直接发送：

var targetQueueAddress = new Uri($"exchange:{builder.Configuration["EventBusSettings:Queue"]}");
var sendEndpoint = await _bus.GetSendEndpoint(targetQueueAddress);
await sendEndpoint.Send(publishAssetRequest);

注：RabbitMQ中MassTransit创建的队列会自动生成同名Exchange，所以用exchange:{队列名}作为地址即可。

方法3：配置消费者时同步绑定发布端点 #

如果消费者和发布端在同一个服务中，可在配置ReceiveEndpoint时，同步绑定消息到该队列的Exchange，确保发布消息自动路由到对应队列：

builder.Services.AddMassTransit(c =>
{
    c.AddConsumer<BlockchainConsumer<PublishAssetRequest>>();
    c.AddConsumer<BlockchainConsumer<PublishAssetMetadataRequest>>();

    c.UsingRabbitMq((context, configurator) =>
    {
        configurator.Host(builder.Configuration["EventBusSettings:HostAddress"]);

        var targetQueue = builder.Configuration["EventBusSettings:Queue"] ?? string.Empty;
        configurator.ReceiveEndpoint(targetQueue, c =>
        {
            c.ConfigureConsumer<BlockchainConsumer<PublishAssetRequest>>(context);
            c.ConfigureConsumer<BlockchainConsumer<PublishAssetMetadataRequest>>(context);
            
            // 绑定消息类型到当前队列的Exchange
            c.Bind<PublishAssetRequest>();
            c.Bind<PublishAssetMetadataRequest>();
        });
    });
});

这种方式无需修改发布代码，消息会自动发送到配置的队列中。

内容的提问来源于stack exchange，提问作者Salva P