使用UseConsumeFilter与Batch Consumers启动时触发NullReferenceException的解决方法

为实现日志记录功能，在MassTransit中配置了消费/发布过滤器后运行正常，但在使用批量消费者（Batch Consumers）的微服务中，注册批量消费者时会抛出NullReferenceException。

相关代码 #

MassTransit 配置代码 #

services.AddMassTransit(config =>
{
    config.AddConsumers(typeof(MyNonBatchConsumer).Assembly);
    
    config.UsingAmazonSqs((context, cfg) =>
    {
        cfg.UsePublishFilter(typeof(LoggingPublishFilter<>), provider);
        cfg.UseConsumeFilter(typeof(LoggingConsumeFilter<>), provider);     

        cfg.ReceiveEndpoint($"{typeof(Startup).Namespace?.Replace(".", "_")}_{environment.EnvironmentName}", receiveEndpointConfig =>
        {
            // 尝试在这里注册过滤器，无效
            receiveEndpointConfig.ConfigureConsumer<MyNonBatchConsumer>(context);
            // 批量消费者注册代码
            receiveEndpointConfig.ConfigureConsumer<MyBatchConsumer>(context);
        });
    }));
});

services.AddScoped(typeof(LoggingPublishFilter<>));
services.AddScoped(typeof(LoggingConsumeFilter<>));

批量处理器代码 #

public class MyBatchConsumerHandler : 
    IIntegrationEventBatchHandler<MyBatchConsumer>
{
    private readonly IMediator _mediator;

    public MyBatchConsumerHandler(IMediator mediator)
    {
        _mediator = mediator;
    }

    public async Task Consume(ConsumeContext<Batch<MyBatchConsumer>> context)
    {
        // 业务逻辑已省略
    }
}

异常详情 #

异常来自ScopedConsumerConsumePipeSpecificationObserver类的ConsumerMessageConfigured<TConsumer, TMessage>(IConsumerMessageConfigurator<TConsumer, TMessage> configurator)方法，原因是批量处理器对应的configurator为null。单独使用消费过滤器或批量消费者均正常，但两者无法协同工作。

问题核心在于批量消费者的注册方式错误，不能用ConfigureConsumer注册批量消费者，需使用MassTransit专属的批量消费者配置API，同时确保过滤器作用域匹配批量消费管道。

正确配置步骤 #

1. 移除错误的批量消费者注册代码：删除receiveEndpointConfig.ConfigureConsumer<MyBatchConsumer>(context);这一行。
2. 使用批量消费者专属配置API：在接收端点中通过Batch方法配置批量消费者，全局消费过滤器会自动应用到批量消费管道，也可针对批量消费单独配置过滤器。
3. 确保批量处理器正确注册：将批量处理器类注册到容器中，保证MassTransit能识别其批量消费逻辑。

修改后的MassTransit配置代码：

services.AddMassTransit(config =>
{
    config.AddConsumers(typeof(MyNonBatchConsumer).Assembly);
    // 注册批量消费者处理器
    config.AddConsumer<MyBatchConsumerHandler>();
    
    config.UsingAmazonSqs((context, cfg) =>
    {
        cfg.UsePublishFilter(typeof(LoggingPublishFilter<>), context);
        cfg.UseConsumeFilter(typeof(LoggingConsumeFilter<>), context);     

        cfg.ReceiveEndpoint($"{typeof(Startup).Namespace?.Replace(".", "_")}_{environment.EnvironmentName}", receiveEndpointConfig =>
        {
            receiveEndpointConfig.ConfigureConsumer<MyNonBatchConsumer>(context);
            
            // 配置批量消费者，可根据业务调整参数
            receiveEndpointConfig.Batch<MyBatchConsumer>(batchConfig =>
            {
                batchConfig.MessageLimit = 10; // 单批最大消息数
                batchConfig.TimeLimit = TimeSpan.FromSeconds(5); // 批量收集超时时间
                // 绑定批量消费者处理器
                batchConfig.Consumer<MyBatchConsumerHandler>(context);
                
                // 如需针对批量消费单独配置过滤器，可取消下方注释
                // batchConfig.UseConsumeFilter(typeof(LoggingConsumeFilter<>), context);
            });
        });
    });
});

services.AddScoped(typeof(LoggingPublishFilter<>));
services.AddScoped(typeof(LoggingConsumeFilter<>));

关键说明 #

• 批量消费者必须通过Batch方法配置，而非ConfigureConsumer，否则MassTransit无法正确创建批量消费管道，会导致configurator为null的异常。
• 全局UseConsumeFilter会自动作用于批量消费者管道，无需重复注册，若需单独调整批量消费的过滤逻辑，可在batchConfig中单独配置。
• 批量处理器类需实现MassTransit的IConsumer<Batch<TMessage>>接口（若自定义了IIntegrationEventBatchHandler接口，需确保其最终适配该接口），并正确注册到依赖注入容器中。

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