多消费者场景下Saga端点配置及EndpointConvention有效性问题

仅添加EndpointConvention.Map<SendNotification>(new Uri("queue:send-notification")); 无法完全满足需求。
这个配置只是为SendNotification类型注册了默认发送端点映射，但没有解决两个核心问题：一是没有约束消息投递模式从Publish切换为Send，二是没有保证消费端队列和映射地址完全匹配，多实例部署下还是可能出现重复消费。

场景1：InMemory传输配置 #

在原有代码基础上做以下调整即可：

• 注册全局端点映射
• 显式定义名为send-notification的接收端点，绑定SendNotification对应的消费者
• 自动配置端点时排除该消费者，避免框架自动生成额外的重复队列
• Saga内投递该消息时，必须调用Send方法，禁止调用Publish

调整后完整代码参考：

// 放在服务注册阶段、传输配置启动前即可
EndpointConvention.Map<SendNotification>(new Uri("queue:send-notification"));

x.UsingInMemory((context, cfg) =>
{
    cfg.UseInMemoryScheduler(out ISchedulerFactory factory); 
    _scheduler = factory.GetScheduler();
    cfg.UseInMemoryOutbox();
    
    // 显式注册目标接收端点
    cfg.ReceiveEndpoint("send-notification", endpointCfg =>
    {
        endpointCfg.ConfigureConsumer<SendNotificationConsumer>(context);
    });

    // 排除已手动注册的消费者，避免重复注册
    cfg.ConfigureEndpoints(context, filter => 
        filter.Exclude<SendNotificationConsumer>());
});

场景2：RabbitMQ传输配置 #

逻辑和InMemory场景完全一致，仅接收端点可按需配置RabbitMQ专属的队列属性（比如持久化、死信规则等），调整后代码参考：

// 同InMemory场景，提前注册全局端点映射
EndpointConvention.Map<SendNotification>(new Uri("queue:send-notification"));

x.UsingRabbitMq((context, cfg) =>
{
    // 显式注册目标接收端点
    cfg.ReceiveEndpoint("send-notification", endpointCfg =>
    {
        endpointCfg.Durable = true; // 按需配置队列持久化等属性
        endpointCfg.ConfigureConsumer<SendNotificationConsumer>(context);
    });

    cfg.UseMessageScheduler(schedulerEndpoint);
    cfg.UseMessageRetry(r => r.Intervals(config.GetSection(RetryIntervalsKey).Get<int[]>()));
    cfg.UseInMemoryOutbox();
    
    // 排除已手动注册的消费者，避免重复注册
    cfg.ConfigureEndpoints(context, filter => 
        filter.Exclude<SendNotificationConsumer>());
});

关键校验点 #

• Saga代码中投递SendNotification必须使用Send方法：EndpointConvention只对Send方法生效，如果你还是调用Publish，消息会走广播逻辑，多实例下每个服务实例都会收到消息，重复发送邮件的问题依然存在。
• 不要完全依赖ConfigureEndpoints自动注册：自动注册生成的队列名默认基于消费者类名，和你指定的queue:send-notification不匹配，会导致消息投递到不存在的队列，或者生成多余的消费队列。
• 部署后验证：启动2个及以上服务实例，触发Saga流程，观察消费日志，send-notification队列的消息只会被其中一个实例消费，不会出现重复处理。

内容的提问来源于stack exchange，提问作者Olivier Quirion