.NET 4.8下MassTransit Azure Service Bus如何添加自定义消息头？

配置失败原因 #

.NET 4.8 仅兼容 MassTransit v7.x 及更早版本，该版本的管道过滤器API与v8+版本存在两处核心差异，导致你按新版文档编写的代码无法生效：

• 新版的全局UseSendFilter()/UseConsumeFilter()顶级入口在v7中不存在，v7要求过滤器必须绑定到特定消息类型的发送/消费管道上，而非全局总线配置层
• 你当前代码中调用的UseSendContext()方法不属于v7版本的有效配置项，缺少分号的语法错误也会直接阻断编译
  同时现有代码存在一处类型匹配偏差：
• 发布调用busControl.Publish(message.Content, CancellationTokenSource.Token)存在逻辑问题：MassTransit会按传入对象的CLR类型匹配管道规则，如果传入的是Content属性对应的字符串值，针对IPublishMessage类型的所有配置完全不会触发

修正后的发布端总线配置 #

var busControl = Bus.Factory.CreateUsingAzureServiceBus(busFactoryConfigurator =>
{
    busFactoryConfigurator.Host(new Uri(Config.ConnectionString), host =>
    {
        host.OperationTimeout = TimeSpan.FromSeconds(30); // 补全原代码中未初始化的TimeSpan值
        // 原有Token Provider配置逻辑
    });

    // 保留原有CorrelationId拓扑配置
    busFactoryConfigurator.SendTopology.UseCorrelationId<IPublishMessage>(x => x.CorrelationId);

    // 针对IPublishMessage类型配置发送管道，自动提取属性写入消息头
    busFactoryConfigurator.Send<IPublishMessage>(sendConfigurator =>
    {
        // 内联执行过滤器，无需额外编写独立Filter类
        sendConfigurator.UseExecute(context =>
        {
            var message = context.Message;
            // 加X-前缀避免与Azure Service Bus、MassTransit系统头冲突
            context.Headers.Set("X-Message-Name", message.Name);
            context.Headers.Set("X-Message-TimeStamp", message.TimeStamp);
            context.Headers.Set("X-Message-Sender", message.Sender);
            context.Headers.Set("X-Message-QueueType", message.QueueType.ToString());
            context.Headers.Set("X-Message-MessageType", message.MessageType.ToString());
            context.Headers.Set("X-Message-ContentType", message.ContentType.ToString());
        });
    });

    // 保留原有Topic实体名配置
    busFactoryConfigurator.Message<IPublishMessage>(topologyConfigurator =>
    {
        topologyConfigurator.SetEntityName(Config.TopicExchange);
    });
});

关键调整说明 #

• 发布调用需修正为传递IPublishMessage实例，而非其Content属性：

  // 正确发布逻辑
  await busControl.Publish<IPublishMessage>(yourMessageInstance, CancellationTokenSource.Token);
  

• 枚举类型头值需显式转为字符串：MassTransit头序列化仅支持基础类型，直接传入枚举值会触发序列化失败

消费端读取自定义头的配置 #

如果需要在消费端读取上述自定义头，对应v7版本的消费端配置无需编写独立ConsumeFilter，直接内联读取即可：

var busControl = Bus.Factory.CreateUsingAzureServiceBus(cfg =>
{
    // 复用现有Host配置
    cfg.ReceiveEndpoint(Config.TargetQueueName, e =>
    {
        e.Consume<IPublishMessage>(context =>
        {
            // 从ConsumeContext中直接读取自定义头
            var sender = context.Headers.Get<string>("X-Message-Sender");
            var timeStamp = context.Headers.Get<DateTime>("X-Message-TimeStamp");
            // 业务消费逻辑
        });
    });
});

独立Filter类的注册方式 #

如果业务逻辑复杂，需要将头赋值逻辑抽为独立类，v7版本的过滤器注册规则为：

1. 实现对应过滤器接口

   public class PublishMessageHeaderFilter : IFilter<SendContext<IPublishMessage>>
   {
       public async Task Send(SendContext<IPublishMessage> context, IPipe<SendContext<IPublishMessage>> next)
       {
           var message = context.Message;
           context.Headers.Set("X-Message-Name", message.Name);
           // 其余头赋值逻辑
           await next.Send(context);
       }
       public void Probe(ProbeContext context) => context.CreateFilterScope("publishMessageHeader");
   }
   

2. 在总线配置中注册

   busFactoryConfigurator.Send<IPublishMessage>(x => x.UseExecuteFilter<PublishMessageHeaderFilter>());
   

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