如何基于CorrelationId过滤MassTransit 8+SQS/SNS的消费者事件？

要实现基于CorrelationId字段过滤事件、指定消费者接收的需求，核心是利用SNS订阅过滤机制，结合MassTransit的配置绑定消费者队列与过滤规则。以下是具体实现步骤：

1. 生产者端：确保CorrelationId被正确传递 #

发布EventData时，需将目标CorrelationId设置到消息头（MassTransit标准头）或消息属性中，确保SNS能识别该字段用于过滤：

方式1：使用MassTransit标准CorrelationId头 #

// 假设EventData是你的通用事件类型
var eventData = new EventData 
{ 
    // 填充事件业务内容
};

// 发布时指定CorrelationId，MassTransit会自动将其转为SNS消息属性
await _bus.Publish(eventData, context =>
{
    context.CorrelationId = Guid.Parse("目标CorrelationId值"); 
});

方式2：使用EventData自身的CorrelationId字段 #

如果EventData本身包含CorrelationId字段，可将其映射为自定义消息头：

await _bus.Publish(eventData, context =>
{
    // 将EventData的CorrelationId添加为自定义消息头
    context.Headers.Set("event-correlation-id", eventData.CorrelationId.ToString());
});

2. 消费者端：配置订阅过滤规则 #

在MassTransit的SQS配置中，为每个消费者的接收端点设置SNS订阅过滤策略，仅让符合CorrelationId条件的消息推送到对应队列。

完整配置示例 #

services.AddMassTransit(x =>
{
    // 注册消费者
    x.AddConsumer<Consumer1>();
    x.AddConsumer<Consumer2>();

    x.UsingAmazonSqs((context, cfg) =>
    {
        cfg.Host("us-east-1"); // 替换为你的AWS区域

        // Consumer1配置：只接收CorrelationId为"consumer1-target-id"的消息
        cfg.ReceiveEndpoint("consumer1-queue", e =>
        {
            e.ConfigureConsumer<Consumer1>(context);
            
            // 设置SNS订阅过滤策略
            e.SubscriptionSettings.FilterPolicy = new Dictionary<string, object>
            {
                { 
                    "correlation-id", // 对应方式1的标准头名称
                    new Dictionary<string, object>
                    {
                        { "StringEquals", "consumer1-target-id" }
                    }
                }
            };
        });

        // Consumer2配置：只接收CorrelationId为"consumer2-target-id"的消息
        cfg.ReceiveEndpoint("consumer2-queue", e =>
        {
            e.ConfigureConsumer<Consumer2>(context);
            
            e.SubscriptionSettings.FilterPolicy = new Dictionary<string, object>
            {
                { 
                    "correlation-id",
                    new Dictionary<string, object>
                    {
                        { "StringEquals", "consumer2-target-id" }
                    }
                }
            };
        });

        // 可选：显式配置EventData对应的共享SNS主题
        cfg.Publish<EventData>(p =>
        {
            p.TopicName = "event-data-shared-topic";
        });
    });
});

灵活过滤规则扩展 #

如果需要匹配多个CorrelationId值，可使用StringListEquals：

e.SubscriptionSettings.FilterPolicy = new Dictionary<string, object>
{
    { 
        "correlation-id",
        new Dictionary<string, object>
        {
            { "StringListEquals", new[] { "id-001", "id-002", "id-003" } }
        }
    }
};

若使用自定义消息头（方式2），只需将过滤规则中的属性名替换为自定义头名称（如event-correlation-id）即可。

3. 原理说明 #

• SNS收到生产者发布的消息后，会根据每个订阅（对应消费者队列）的过滤规则判断是否将消息推送到队列。
• MassTransit通过SubscriptionSettings.FilterPolicy将过滤规则传递给SNS，无需手动在AWS控制台配置。
• 只有符合CorrelationId条件的消息才会被转发到对应的消费者队列，消费者只会处理目标消息。

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