MassTransit Kafka消息反序列化异常拦截方案求助

近期使用MassTransit 8.0.13对接Kafka，遇到如下需求：客户端发送非法JSON格式消息到Kafka时，服务端需拦截反序列化异常，并向指定Kafka主题发送格式化响应消息。

最初尝试通过过滤器和中间件实现，但该异常发生在消息进入Consumer之前，无法被现有逻辑捕获；后续尝试编写自定义反序列化器，计划在其中调用TopicProducer发送响应，但在配置中通过ServiceCollection.BuildServiceProvider()获取服务时，触发错误：The ReceivePipeConfiguration can only be used once.。

需要合理的实现方案，正确拦截MassTransit Kafka消息的反序列化异常并执行后续操作。

方案一：利用MassTransit内置错误回调机制（推荐） #

MassTransit的Kafka端点支持通过ConfigureErrorCallback直接捕获反序列化异常，无需自定义反序列化器，是官方推荐的处理方式：

• 在TopicEndpoint配置中添加错误回调，回调内可直接获取服务上下文、异常信息及原始消息，进而调用生产者发送响应。
• 无需手动构建ServiceProvider，直接通过上下文获取所需服务（如Producer）。

方案二：正确实现自定义反序列化器（依赖注入） #

若必须使用自定义反序列化器，需避免在配置阶段手动构建ServiceProvider，而是通过MassTransit提供的上下文获取服务：

• 不要在TopicEndpoint配置中调用services.BuildServiceProvider()，改用ctx（IKafkaRiderContext）的GetService<T>或Resolve<T>方法获取依赖。
• 将自定义反序列化器注册到依赖注入容器，确保其依赖能被正确解析。

方案一（错误回调实现） #

public static void AddMassTransitConfiguration(
    this IServiceCollection services, 
    IConfiguration configuration)
{
    services.AddMassTransit(busCfg =>
    {
        busCfg.UsingInMemory((context, config) =>
        {
            config.ConfigureEndpoints(context);
        });

        busCfg.AddRider(rider =>
        {
            rider.AddConsumer<OrderConsumer>(); 
            rider.AddConsumer<OrderProcessingResultConsumer>();

            rider.AddProducer<OrderForProcessingMessage>(
                configuration
                    .GetSection("OrderForProcessingProducer")
                    .GetSection("Topic").Value,
                (context, cfg) =>
                {
                    cfg.SetValueSerializer(
                        new MassTransitJsonSerializer<OrderForProcessingMessage>());
                });
           
            rider.AddProducer<OrderResultMessage>(
                configuration.GetSection("OrderResultProducer").GetSection("Topic").Value,
                (context, cfg) =>
                {
                    cfg.SetValueSerializer(new MassTransitJsonSerializer<OrderResultMessage>()); 
                });

            var isDevelopment = Environment.GetEnvironmentVariable("ASPNETCORE_ENVIRONMENT") !=
                                Environments.Development;

            Action<IKafkaHostConfigurator>? kafkaHostConfigurator = isDevelopment ? hostConfig =>
                {
                    hostConfig.UseSasl(saslConf =>
                    {
                        saslConf.Mechanism = SaslMechanism.ScramSha256;
                        saslConf.SecurityProtocol = SecurityProtocol.SaslPlaintext;
                        saslConf.Username = configuration.GetSection("Kafka")
                            .GetSection("User").Value;
                        saslConf.Password = configuration.GetSection("Kafka")
                            .GetSection("Password").Value;
                    });
                }
                : null;

            rider.UsingKafka((ctx, kafkaCfg) =>
            {
                kafkaCfg.Host(configuration.GetSection("Kafka").GetSection("Host").Value, kafkaHostConfigurator);
                
                // 处理OrderMessage主题的反序列化异常
                kafkaCfg.TopicEndpoint<OrderMessage>(
                    configuration.GetSection("OrderConsumer").GetSection("Topic").Value,
                    configuration.GetSection("OrderConsumer").GetSection("ConsumerGroup").Value,
                    endpointCfg =>
                    {
                        endpointCfg.AutoOffsetReset = AutoOffsetReset.Earliest;
                        endpointCfg.SetValueDeserializer(new MassTransitJsonSerializer<OrderMessage>());
                        endpointCfg.ConfigureConsumer<OrderConsumer>(ctx);

                        // 配置错误回调，捕获反序列化异常
                        endpointCfg.ConfigureErrorCallback(async (errorContext, exception) =>
                        {
                            // 获取OrderResultMessage的生产者
                            var resultProducer = errorContext.GetProducer<OrderResultMessage>();
                            
                            // 构造错误响应消息
                            var errorResponse = new OrderResultMessage
                            {
                                ErrorCode = "DESERIALIZATION_FAILED",
                                ErrorMessage = $"反序列化失败: {exception.Message}",
                                OriginalMessage = errorContext.RawMessage != null 
                                    ? Encoding.UTF8.GetString(errorContext.RawMessage) 
                                    : "无法获取原始消息"
                            };

                            // 发送响应到指定主题
                            await resultProducer.Produce(errorResponse);

                            // 返回Ack确认消息已处理，避免重复消费；若需重试可返回Retry
                            return ConsumerResult.Ack;
                        });
                    }
                );

                kafkaCfg.TopicEndpoint<OrderProcessingResultMessage>(
                    configuration.GetSection("OrderProcessingResultConsumer").GetSection("Topic").Value,
                    configuration.GetSection("OrderProcessingResultConsumer").GetSection("ConsumerGroup").Value,
                    endpointCfg => 
                    {
                        endpointCfg.AutoOffsetReset = AutoOffsetReset.Earliest;
                        endpointCfg.SetValueDeserializer(new MassTransitJsonDeserializer<OrderProcessingResultMessage>());
                        endpointCfg.ConfigureConsumer<OrderProcessingResultConsumer>(ctx);
                    }
                );
            });
        });
    });
}

方案二（自定义反序列化器正确注入） #

首先注册自定义反序列化器到容器：

services.AddScoped<CustromMassTransitJsonDeserializer>();

然后在配置中通过上下文解析：

kafkaCfg.TopicEndpoint<OrderMessage>(
    configuration.GetSection("OrderConsumer").GetSection("Topic").Value,
    configuration.GetSection("OrderConsumer").GetSection("ConsumerGroup").Value,
    endpointCfg =>
    {
        endpointCfg.AutoOffsetReset = AutoOffsetReset.Earliest;
        // 通过上下文解析自定义反序列化器，自动注入依赖
        var deserializer = ctx.Resolve<CustromMassTransitJsonDeserializer>();
        endpointCfg.SetValueDeserializer(deserializer);
        endpointCfg.ConfigureConsumer<OrderConsumer>(ctx);
    }
);

自定义反序列化器示例：

public class CustromMassTransitJsonDeserializer : IDeserializer<OrderMessage>
{
    private readonly IBeeKeeperIntegrationService _beeKeeperService;
    private readonly IProducer<OrderResultMessage> _resultProducer;

    // 通过构造函数注入所需服务
    public CustromMassTransitJsonDeserializer(
        IBeeKeeperIntegrationService beeKeeperService,
        IProducer<OrderResultMessage> resultProducer)
    {
        _beeKeeperService = beeKeeperService;
        _resultProducer = resultProducer;
    }

    public async Task<OrderMessage> DeserializeAsync(
        ReadOnlyMemory<byte> data, 
        bool isKey, 
        MessageContext context)
    {
        try
        {
            // 执行反序列化逻辑
            var json = Encoding.UTF8.GetString(data.Span);
            return JsonSerializer.Deserialize<OrderMessage>(json);
        }
        catch (Exception ex)
        {
            // 构造错误响应并发送
            var errorResponse = new OrderResultMessage
            {
                ErrorMessage = $"反序列化失败: {ex.Message}",
                OriginalMessage = Encoding.UTF8.GetString(data.Span)
            };
            await _resultProducer.Produce(errorResponse);

            // 可根据需求抛出异常或返回默认值（注意：抛出异常仍会触发MassTransit的错误处理）
            throw;
        }
    }
}

• 禁止在MassTransit配置阶段调用ServiceCollection.BuildServiceProvider()，这会导致容器重复构建，触发The ReceivePipeConfiguration can only be used once.错误。
• 使用错误回调时，errorContext.RawMessage可获取原始未序列化的消息字节，便于排查问题。
• 若选择自定义反序列化器，需确保其依赖（如Producer、业务服务）已正确注册到依赖注入容器。

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