如何在MassTransit中以ActiveMQ为传输创建持久化消费者？

问题背景 #

正在为大型应用搭建基础架构，希望配置准确。熟悉Azure Service Bus和RabbitMQ，但对ActiveMQ与MassTransit的组合不熟悉，虽二者搭配表现优秀，但相关文档及示例项目较少。需要配置接收端点使其持久化，让从Topic发布的消息存储到磁盘，已知配置需在接收端靠近传输配置（ActiveMQ Artemis），当前注册消费者的代码如下：

services.AddMassTransit(busConfigurator =>
{
    //var entryAssembly = Assembly.GetExecutingAssembly();
    //busConfigurator.AddConsumers(entryAssembly);

    busConfigurator.AddConsumer<JournalHasBeenViewedMessageConsumer>(typeof(JournalHasBeenViewedMessageConsumer), (consumeConfig) =>
    {
    });

    busConfigurator.UsingActiveMq((context, cfg) =>
    {
        cfg.Host("localhost", 61616, h =>
        {
            h.Username("admin");
            h.Password("admin");
        });

        cfg.EnableArtemisCompatibility();

        // Here right?

        cfg.ConfigureEndpoints(context);
    });
});

配置方案 #

根据MassTransit官方文档，Durable属性控制消息持久化行为，相关说明如下：

MassTransit包含多个控制接收端点行为的配置选项...

属性	类型	说明
Durable	bool	如果为true，消息在被确认前会持久化到磁盘（默认值：true）

虽然Durable默认值已经是true，但显式配置能明确意图，避免后续默认值变更影响。以下提供三种配置方式：

1. 全局配置所有接收端点

在UsingActiveMq块中，通过ReceiveEndpointDefaults统一设置所有端点的持久化：

services.AddMassTransit(busConfigurator =>
{
    busConfigurator.AddConsumer<JournalHasBeenViewedMessageConsumer>();

    busConfigurator.UsingActiveMq((context, cfg) =>
    {
        cfg.Host("localhost", 61616, h =>
        {
            h.Username("admin");
            h.Password("admin");
        });

        cfg.EnableArtemisCompatibility();

        // 全局配置所有接收端点的持久化（显式设置，默认已为true）
        cfg.ReceiveEndpointDefaults.Configure(e =>
        {
            e.Durable = true;
        });

        cfg.ConfigureEndpoints(context);
    });
});

2. 针对单个消费者端点配置

在AddConsumer时，直接为该消费者对应的端点配置持久化：

services.AddMassTransit(busConfigurator =>
{
    busConfigurator.AddConsumer<JournalHasBeenViewedMessageConsumer>(configure =>
    {
        // 为当前消费者的接收端点设置持久化
        configure.Endpoint(e =>
        {
            e.Durable = true;
        });
    });

    busConfigurator.UsingActiveMq((context, cfg) =>
    {
        cfg.Host("localhost", 61616, h =>
        {
            h.Username("admin");
            h.Password("admin");
        });

        cfg.EnableArtemisCompatibility();

        cfg.ConfigureEndpoints(context);
    });
});

3. 直接配置特定命名端点

如果需要自定义端点名称，可在UsingActiveMq中直接创建并配置接收端点：

services.AddMassTransit(busConfigurator =>
{
    busConfigurator.AddConsumer<JournalHasBeenViewedMessageConsumer>();

    busConfigurator.UsingActiveMq((context, cfg) =>
    {
        cfg.Host("localhost", 61616, h =>
        {
            h.Username("admin");
            h.Password("admin");
        });

        cfg.EnableArtemisCompatibility();

        // 配置指定名称的接收端点，并启用持久化
        cfg.ReceiveEndpoint("journal-has-been-viewed", e =>
        {
            e.Durable = true;
            e.ConfigureConsumer<JournalHasBeenViewedMessageConsumer>(context);
        });

        // 若有其他消费者，可继续添加或使用ConfigureEndpoints处理
        // cfg.ConfigureEndpoints(context);
    });
});

配置说明 #

• 开启Durable后，ActiveMQ Artemis会将消息写入磁盘，即使服务重启，未确认的消息也不会丢失
• 针对Topic场景，消费者绑定的订阅队列会被标记为持久化，确保从Topic转发的消息被持久存储
• 显式设置Durable = true是最佳实践，能让配置意图更清晰，避免依赖默认值带来的潜在风险

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