MassTransit配置消费者时双队列生成及多播队列配置问题

解决方案 #

1. 清除无用的Anycast队列 #

MassTransit默认会给每个消息类型生成一个发送端点队列（用于支持Send操作），但咱们用的是发布/订阅模式，完全不需要这玩意儿。直接针对事件类型禁用发送拓扑即可：

// 禁用指定事件的发送拓扑，不会生成对应的Anycast队列
cfg.SendTopology.ConfigureMessageTopology<JournalHasBeenViewedMessage>(false);

如果有多个事件类型，要么重复上述代码，要么写个小工具类批量处理所有事件类型。

2. 让自定义命名队列变为Multicast类型 #

使用自定义队列命名器后，默认生成的队列会变成Anycast，此时需要手动给队列添加Multicast属性。可以在ConfigureEndpoints的回调里统一配置：

完整配置代码 #

configurator.UsingActiveMq((context, cfg) =>
{
    var configuration = new ConfigurationHostSettings(new Uri($"activemq://{options.Host}"));

    configuration.TransportOptions.Add("transport.useInactivityMonitor", "false");
    configuration.Username = options.User;
    configuration.Password = options.Password;

    cfg.Host(configuration);
    cfg.EnableArtemisCompatibility();

    // 禁用事件的发送拓扑，避免生成无用队列
    cfg.SendTopology.ConfigureMessageTopology<JournalHasBeenViewedMessage>(false);

    var queueNameFormatter = new RgsQueueNameFormatter();

    // 自定义每个消费者端点的配置
    cfg.ConfigureEndpoints(context, endpointConfigurator =>
    {
        // 应用自定义命名规则
        endpointConfigurator.SetQueueNameFormatter(queueNameFormatter);
        
        // 强转成ActiveMQ的端点配置器，设置Multicast属性
        if (endpointConfigurator is ActiveMqReceiveEndpointConfigurator activeMqEndpoint)
        {
            // 给队列和地址都加上multicast参数，让Artemis识别为多播队列
            activeMqEndpoint.SetQueueArgument("multicast", "true");
            activeMqEndpoint.SetAddressArgument("multicast", "true");
        }
    });
});

备选方案：手动注册消费者端点 #

如果自动配置仍有问题，直接手动注册消费者端点，完全掌控队列创建和订阅逻辑：

cfg.ReceiveEndpoint(queueNameFormatter.FormatEndpointName(typeof(JournalHasBeenViewedConsumer)), e =>
{
    e.Consumer<JournalHasBeenViewedConsumer>(context);
    // 设置多播属性
    e.SetQueueArgument("multicast", "true");
    e.SetAddressArgument("multicast", "true");
    // 订阅对应的虚拟主题
    e.Subscribe("VirtualTopic.JournalHasBeenViewed");
});

注意事项 #

• 确保RgsQueueNameFormatter生成的队列名称和Artemis的虚拟主题规则匹配，比如包含VirtualTopic.前缀（或与订阅的主题名称对应）。
• 禁用发送拓扑是核心步骤，否则那个无用的同名队列还会自动生成。

内容的提问来源于stack exchange，提问作者Magnus Backeus