MassTransit中自定义过滤器致主题和订阅名称前缀丢失问题

问题描述 #

使用MassTransit 8结合Amazon SQS/SNS，部署在Windows Docker容器中。为消费者添加自定义过滤器后，主题和订阅名称丢失了预设的范围前缀，但队列名称始终保持正确前缀。

复现代码 #

自定义过滤器代码 #

public sealed class CustomConsumeFilter<T> : IFilter<ConsumeContext<T>> where T : class
{
    public async Task Send(ConsumeContext<T> context, IPipe<ConsumeContext<T>> next)
    {
        await next.Send(context);
    }

    public void Probe(ProbeContext context)
    {
    }
}

消费者代码 #

public class TestModelEventsConsumer : IConsumer<TestModel>
{
    public Task Consume(ConsumeContext<TestModel> context)
    {
        return Task.CompletedTask;
    }
}

MassTransit初始化代码 #

var builder = Host.CreateApplicationBuilder(args);

var configuration = builder.Configuration;

builder.Services
       .Configure<AmazonSqsTransportOptions>(configuration.GetSection("MassTransit:Transports:AmazonSqs"))
       .AddMassTransit(bus =>
       {
           bus.AddConsumer<TestModelEventsConsumer>();            

           bus.UsingAmazonSqs((context, sqs) =>
           {
               sqs.UseFilter(new CustomConsumeFilter<TestModel>()); // 问题触发点

               var entityNameFormatter = new MessageNameFormatterEntityNameFormatter(new AmazonSqsMessageNameFormatter());

               var scope = "test-prefix-";
           
               sqs.MessageTopology.SetEntityNameFormatter(new PrefixEntityNameFormatter(entityNameFormatter, scope));

               sqs.ConfigureEndpoints(context, new KebabCaseEndpointNameFormatter(scope));
           });
       });

builder.Build().Run();

现象对比 #

添加过滤器时的资源名称 #

• 队列：test-prefix-test-model-events - 正确（带前缀）
• 主题：MyApplication_Services-TestModel - 错误（无前缀）
• 订阅ARN：arn:aws:sns:eu-south-1:111111111111:MyApplication_Services-TestModel:66904f2a-fc68-4652-b038-ef2d6157b7c9 - 错误（无前缀）

移除过滤器时的资源名称 #

• 队列：test-prefix-test-model-events - 正确（带前缀）
• 主题：test-prefix-MyApplication_Services-TestModel - 正确（带前缀）
• 订阅ARN：arn:aws:sns:eu-south-1:111111111111:test-prefix-MyApplication_Services-TestModel:b8eb1111-65ca-4833-9b3f-29df2b33a8c9 - 正确（带前缀）

问题原因 #

这并非MassTransit的Bug，而是配置顺序触发的拓扑初始化时机问题：

• 当调用sqs.UseFilter(new CustomConsumeFilter<TestModel>())时，MassTransit会立即为TestModel类型构建消息拓扑，此时EntityNameFormatter尚未设置（后续代码才调用sqs.MessageTopology.SetEntityNameFormatter），因此主题名称未应用前缀。
• 队列名称由ConfigureEndpoints传入的KebabCaseEndpointNameFormatter控制，该配置在过滤器之后执行，因此不受影响。

解决方法 #

方法1：调整配置顺序，先设置拓扑再添加过滤器 #

将EntityNameFormatter的设置移到过滤器之前，确保拓扑初始化时已应用前缀规则：

bus.UsingAmazonSqs((context, sqs) =>
{
    var scope = "test-prefix-";
    var entityNameFormatter = new MessageNameFormatterEntityNameFormatter(new AmazonSqsMessageNameFormatter());
    sqs.MessageTopology.SetEntityNameFormatter(new PrefixEntityNameFormatter(entityNameFormatter, scope));

    sqs.UseFilter(new CustomConsumeFilter<TestModel>()); // 移到拓扑设置之后

    sqs.ConfigureEndpoints(context, new KebabCaseEndpointNameFormatter(scope));
});

方法2：将过滤器绑定到消费者级别（推荐） #

不在总线全局添加特定消息类型的过滤器，而是在消费者配置中绑定，避免提前触发拓扑初始化：

bus.AddConsumer<TestModelEventsConsumer>(cfg =>
{
    cfg.UseFilter(new CustomConsumeFilter<TestModel>()); // 仅对该消费者应用过滤器
});

bus.UsingAmazonSqs((context, sqs) =>
{
    var scope = "test-prefix-";
    var entityNameFormatter = new MessageNameFormatterEntityNameFormatter(new AmazonSqsMessageNameFormatter());
    sqs.MessageTopology.SetEntityNameFormatter(new PrefixEntityNameFormatter(entityNameFormatter, scope));

    sqs.ConfigureEndpoints(context, new KebabCaseEndpointNameFormatter(scope));
});

方法3：使用全局通用过滤器 #

如果需要对所有消息应用过滤器，实现针对ConsumeContext的通用过滤器，而非特定消息类型：

public sealed class GlobalConsumeFilter : IFilter<ConsumeContext>
{
    public async Task Send(ConsumeContext context, IPipe<ConsumeContext> next)
    {
        await next.Send(context);
    }

    public void Probe(ProbeContext context)
    {
    }
}

// 总线配置中使用
sqs.UseFilter(new GlobalConsumeFilter());

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