如何在MassTransit中为批量消费者添加单消息消费过滤器

核心配置思路 #

在MassTransit中，批量消费的流程是先收集单条消息再组装成批次，因此只需在批量消费配置前对单条消息应用过滤器，即可实现每条消息先经过过滤，再进入批次等待批量处理。同时通过依赖注入解决过滤器的服务获取问题，无需手动调用GetPayload<IServiceProvider>。

步骤1：修改接收端点配置 #

将原有的单条消费者配置替换为批量消费配置，确保过滤器在批量配置前添加：

cfg.ReceiveEndpoint("BillingTransaction", e =>
{
    e.UseMessageRetry(r => r.Interval(defaultRetryCount, defaultRetryPeriod));

    // 对每条单条ITransactionEvent应用过滤器，顺序执行
    e.UseConsumeFilter(typeof(CacheTransactionFilter<>), context);
    e.UseConsumeFilter(typeof(TransactionValidationFilter<>), context);
    e.UseConsumeFilter(typeof(TransactionDispatchingFilter<>), context);

    // 配置批量消费规则，指定批次大小和超时
    e.ConfigureBatch<ITransactionEvent>(context, batch =>
    {
        batch.MessageLimit = 10; // 单批次最大消息数量
        batch.TimeLimit = TimeSpan.FromSeconds(5); // 批次超时时间（到点即使未达数量也触发消费）
        batch.ConfigureConsumer<TransactionEventConsumer>(context);
    });

    EndpointConvention.Map<ITransactionEvent>(e.InputAddress);
});

步骤2：确保过滤器支持依赖注入 #

过滤器通过构造函数注入所需服务，无需手动获取ServiceProvider，示例如下：

public class CacheTransactionFilter<T> : IFilter<ConsumeContext<T>> 
    where T : class, ITransactionEvent
{
    private readonly ICacheService _cacheService;

    // 构造函数注入依赖服务
    public CacheTransactionFilter(ICacheService cacheService)
    {
        _cacheService = cacheService;
    }

    public async Task Send(ConsumeContext<T> context, IPipe<ConsumeContext<T>> next)
    {
        // 对单条消息执行过滤逻辑
        await _cacheService.CacheTransaction(context.Message);
        
        // 传递消息到下一个过滤器/流程
        await next.Send(context);
    }

    public void Probe(ProbeContext context)
    {
        context.CreateFilterScope("cache-transaction");
    }
}

步骤3：注册过滤器到DI容器 #

将所有过滤器注册到依赖注入容器，MassTransit会自动解析实例并注入依赖：

services.AddScoped(typeof(CacheTransactionFilter<>));
services.AddScoped(typeof(TransactionValidationFilter<>));
services.AddScoped(typeof(TransactionDispatchingFilter<>));

步骤4：实现批量消费者 #

修改TransactionEventConsumer以支持批量消费，示例如下：

public class TransactionEventConsumer : IConsumer<Batch<ITransactionEvent>>
{
    public async Task Consume(ConsumeContext<Batch<ITransactionEvent>> context)
    {
        // 处理已过滤后的批量消息
        foreach (var messageContext in context.Message)
        {
            var transaction = messageContext.Message;
            // 批量处理逻辑
        }
    }
}

关键说明 #

1. 过滤器执行顺序：添加的UseConsumeFilter会按照代码顺序依次对每条单条消息执行过滤逻辑，只有通过所有过滤器的消息才会进入批次。
2. DI容器兼容：使用UseConsumeFilter(typeof(TFilter), context)的方式，MassTransit会自动从DI容器中获取过滤器实例，无需手动处理服务提供器。

内容的提问来源于stack exchange，提问作者Максим Белов