如何为Azure服务编写MassTransit JSON反序列化器解决消息解析失败问题

这个问题我之前帮人排查过，核心原因很明确——MassTransit默认只识别带有自己专属Content-Type的消息（比如application/vnd.masstransit+json），而Event Grid发布的消息用的是标准的application/json; charset=utf-8，所以MassTransit找不到对应的反序列化器来处理这条消息。给你两个可行的解决方案，你可以根据自己的需求选择：

解决方案1：让MassTransit直接支持标准application/json类型 #

这是最简洁的方案，只需要在MassTransit的配置中，让Json反序列化器兼容标准的application/json Content-Type即可。修改你现有AddBus方法里的配置：

services.AddSingleton(provider => Bus.Factory.CreateUsingAzureServiceBus(cfg =>
{
    // 保留原有的Host配置
    var keyName = "RootManageSharedAccessKey";
    var busName = configuration["ServiceBus:Name"];
    var secret = configuration["ServiceBus:Secret"];
    var host = cfg.Host(
        $"Endpoint=sb://{busName}.servicebus.windows.net/;SharedAccessKeyName={keyName};SharedAccessKey={secret}", 
        z => TokenProvider.CreateSharedAccessSignatureTokenProvider(keyName, secret));

    // 配置Json序列化器（保留你原有的自定义转换器）
    cfg.ConfigureJsonSerializer(settings =>
    {
        settings.Converters.Add(new InterfaceConverter());
        return settings;
    });

    // 关键配置：让Json反序列化器支持application/json类型
    cfg.ConfigureJsonDeserializer(settings =>
    {
        settings.Converters.Add(new InterfaceConverter());
        settings.TypeNameHandling = TypeNameHandling.Auto; // 处理接口/多态类型的必备配置
        return settings;
    }, contentType => 
        contentType == "application/json" || contentType.StartsWith("application/json;"));

    // 保留原有的日志、消费者配置
    cfg.UseExtensionsLogging(provider.GetService<ILoggerFactory>());
    cfg.ReceiveEndpoint(host, configuration["ServiceBus:Topic"], e =>
    {
        e.Consumer<NomeEmailChangeConsumer>(provider);
    });
}));

这里的核心是ConfigureJsonDeserializer的第二个参数，它指定了哪些Content-Type应该用这个反序列化器处理——我们把标准的application/json（包括带charset的变种）都加了进去。

解决方案2：自定义反序列化器（处理复杂消息结构） #

如果Event Grid的消息是封装在EventGridEvent结构里（比如带Id、EventType、Data这些字段），而你的消费者需要直接处理Data里的业务对象，那可以自定义一个反序列化器来解析这个结构：

首先创建自定义反序列化器：

public class EventGridJsonDeserializer : IMessageDeserializer
{
    private readonly JsonSerializerSettings _settings;

    public EventGridJsonDeserializer(JsonSerializerSettings settings)
    {
        _settings = settings;
    }

    public ContentType ContentType => new ContentType("application/json");

    public ConsumeContext Deserialize(ReceiveContext receiveContext)
    {
        using var stream = receiveContext.Body.AsStream();
        using var reader = new StreamReader(stream);
        using var jsonReader = new JsonTextReader(reader);

        // 先解析EventGrid的外层结构
        var serializer = JsonSerializer.Create(_settings);
        var eventGridEvent = serializer.Deserialize<EventGridEvent>(jsonReader);

        // 把Data字段转换成你的业务对象（比如NomeEmailChange）
        var businessMessage = JsonConvert.DeserializeObject<NomeEmailChange>(
            JsonConvert.SerializeObject(eventGridEvent.Data), 
            _settings);

        // 构造MassTransit需要的ConsumeContext
        var headers = receiveContext.TransportHeaders.ToDictionary(kv => kv.Key, kv => kv.Value);
        return new JsonConsumeContext(receiveContext, businessMessage, headers, _settings);
    }
}

然后在MassTransit配置中注册这个反序列化器：

services.AddSingleton(provider => Bus.Factory.CreateUsingAzureServiceBus(cfg =>
{
    // 保留原有的Host配置...

    // 初始化Json配置并注册自定义反序列化器
    var jsonSettings = new JsonSerializerSettings
    {
        Converters.Add(new InterfaceConverter()),
        TypeNameHandling = TypeNameHandling.Auto
    };
    cfg.AddMessageDeserializer(
        new ContentType("application/json"), 
        () => new EventGridJsonDeserializer(jsonSettings));

    // 保留原有的日志、消费者配置...
}));

额外注意事项 #

• 确保你的NomeEmailChangeConsumer接收的消息类型和最终反序列化出来的对象类型一致；
• 如果Event Grid直接发送的就是NomeEmailChange的JSON（没有外层的EventGridEvent结构），那么方案1就能直接生效；如果是封装结构，方案2更合适。

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