MassTransit自定义反序列化在Azure Service Bus订阅端点失效求助

对接Azure Service Bus时，外部应用发来的自定义JSON消息（无MassTransit默认要求的messageType和message包裹结构）无法被消费者接收。尝试使用ClearSerialization和UseRawJsonSerializer配置未生效，以下是问题细节和解决方案：

问题场景 #

外部消息格式（无法被处理）：

{
    "planId": "1",
    "description": "Startup Event"
}

MassTransit默认支持的消息格式：

{
    "messageType": [
        "urn:message.Enterprise.TM.Core.Events.ServiceBus:EventPlanCreated"
    ],
    "message": {
        "planId": "1",
        "description": "Startup Event"
    }
}

现有代码问题 #

消费者与实体类 #

实体类EventPlanCreated的字段名与外部消息不匹配（消息中是planId，实体类中是EventId），这会直接导致反序列化失败：

public class ServiceBusEventConsumer : IConsumer<EventPlanCreated>
{
    private readonly IServiceBusEventProcessorFactory _serviceBusEventProcessorFactory;

    public ServiceBusEventConsumer(IServiceBusEventProcessorFactory serviceBusEventProcessorFactory)
    {
        _serviceBusEventProcessorFactory = serviceBusEventProcessorFactory;
    }

    public Task Consume(ConsumeContext<EventPlanCreated> context)
    {
        // 业务逻辑
        return Task.CompletedTask;
    }
}

// 字段名不匹配的实体类
public class EventPlanCreated
{
    public string EventId{ get; set; } // 对应消息中的planId，字段名错误
    public string Description{ get; set; }
}

配置问题 #

序列化配置未正确处理消息类型映射——外部消息没有messageType头，MassTransit无法自动识别要反序列化的目标类型。

解决方案 #

1. 修正实体类字段映射 #

确保实体类字段与消息字段完全匹配，可使用JsonPropertyName特性指定JSON字段名：

using System.Text.Json.Serialization;

public class EventPlanCreated
{
    [JsonPropertyName("planId")]
    public string PlanId { get; set; }

    [JsonPropertyName("description")]
    public string Description { get; set; }
}

2. 正确配置RawJsonSerializer与消息类型解析 #

在MassTransit配置中，需要明确指定消息类型解析逻辑，告诉MassTransit将接收到的原始JSON消息映射到EventPlanCreated类型：

builder.Services.AddMassTransit(x =>
{
    x.AddConsumer<ServiceBusEventConsumer>();

    x.UsingAzureServiceBus((context, cfg) =>
    {
        cfg.Host(builder.Configuration["ServiceBusSettings:ConnectionString"]);

        // 全局清除默认序列化，启用RawJsonSerializer
        cfg.ClearSerialization();
        cfg.UseRawJsonSerializer();

        cfg.SubscriptionEndpoint<EventPlanCreated>("tmsuscription", e =>
        {
            e.ConfigureConsumeTopology = false;
            e.ConfigureConsumer<ServiceBusEventConsumer>(context);

            // 配置消息类型解析，适配无messageType头的外部消息
            e.UseRawJsonSerializer(options =>
            {
                options.MessageTypeResolver = message => 
                {
                    // 指定当前消息对应的目标类型URN
                    return new[] { typeof(EventPlanCreated).GetUrn() };
                };
            });
        });
    });
});

3. 可选：直接接收原始消息手动反序列化 #

如果上述配置仍有问题，可改为接收原始消息后手动反序列化：

// 修改消费者为接收RawMessage
public class ServiceBusEventConsumer : IConsumer<RawMessage>
{
    private readonly IServiceBusEventProcessorFactory _serviceBusEventProcessorFactory;
    private readonly JsonSerializerOptions _jsonOptions;

    public ServiceBusEventConsumer(IServiceBusEventProcessorFactory serviceBusEventProcessorFactory)
    {
        _serviceBusEventProcessorFactory = serviceBusEventProcessorFactory;
        _jsonOptions = new JsonSerializerOptions
        {
            PropertyNameCaseInsensitive = true
        };
    }

    public async Task Consume(ConsumeContext<RawMessage> context)
    {
        // 手动反序列化原始消息
        var message = JsonSerializer.Deserialize<EventPlanCreated>(context.Message.Body, _jsonOptions);
        
        // 后续业务逻辑
        // var strategy = _serviceBusEventProcessorFactory.GetServiceBusEventHandlerStrategy(...);
        // await strategy.HandleEventAsync(message);
    }
}

// 配置中改为订阅RawMessage
builder.Services.AddMassTransit(x =>
{
    x.AddConsumer<ServiceBusEventConsumer>();

    x.UsingAzureServiceBus((context, cfg) =>
    {
        cfg.Host(builder.Configuration["ServiceBusSettings:ConnectionString"]);

        cfg.SubscriptionEndpoint<RawMessage>("tmsuscription", e =>
        {
            e.ConfigureConsumeTopology = false;
            e.ConfigureConsumer<ServiceBusEventConsumer>(context);
        });
    });
});

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