NServiceBus发送消息时如何访问RabbitMQ的IBasicProperties

之前Behavior修改不生效的原因 #

NServiceBus消息管道分为逻辑处理层、传输层两个独立阶段，普通Behavior运行在逻辑处理层，此时RabbitMQ传输层还未创建IBasicProperties实例，逻辑层设置的自定义属性不会同步到最终的RabbitMQ客户端属性对象，后续传输层初始化IBasicProperties时会覆盖逻辑层的相关配置，导致修改不生效。

可行实现方案 #

根据使用场景，可以选择以下两种经过验证的实现方式，两种方式的自定义逻辑都会在BasicPublish调用前最后执行，不会被框架覆盖：

• 方案1：单条消息定向配置（适合仅给发往第三方的特定消息加属性）
  NServiceBus.RabbitMQ 传输包原生提供了IBasicProperties修改入口，在构造发送选项时直接注入修改委托即可，不需要额外注册管道组件：

  // 构造发送配置
  var sendOptions = new SendOptions();
  // 指定消息发送到第三方服务对应的队列
  sendOptions.SetDestination("ThirdParty_Target_Queue");
  // 注入属性修改逻辑
  sendOptions.SetRabbitMqPropertiesModification(basicProperties =>
  {
      // 按第三方要求设置属性，如果要求是BasicProperties内置的ContentEncoding为utf-8，用下一行
      basicProperties.ContentEncoding = "utf-8";
      // 如果第三方要求是在消息头中添加encoding_type字段，替换为下一行即可
      // basicProperties.Headers["encoding_type"] = "utf-8";
  });
  // 执行消息发送
  await endpoint.Send(yourThirdPartyMessage, sendOptions);
  

  该扩展方法位于NServiceBus命名空间下，引入对应命名空间即可直接调用。
• 方案2：全局自动配置（适合所有发往指定队列的消息统一加属性）
  如果需要批量给发往第三方队列的所有消息自动配置属性，可以实现自定义消息派发器，在传输层拦截派发操作统一修改：
  第一步，在端点配置时注册自定义派发器：

  var endpointConfig = new EndpointConfiguration("YourServiceEndpoint");
  var rabbitMqTransport = endpointConfig.UseTransport<RabbitMQTransport>();
  // 注册自定义派发拦截器
  rabbitMqTransport.UseDispatcher<ThirdPartyQueuePropertyDispatcher>();
  

  第二步，实现自定义派发器逻辑：

  public class ThirdPartyQueuePropertyDispatcher : IDispatchMessages
  {
      private readonly IDispatchMessages _innerCoreDispatcher;
  
      // 框架会自动注入原生派发器实例
      public ThirdPartyQueuePropertyDispatcher(IDispatchMessages innerCoreDispatcher)
      {
          _innerCoreDispatcher = innerCoreDispatcher;
      }
  
      public async Task Dispatch(TransportOperations transportOps, TransportTransaction transaction, CancellationToken cancellationToken = default)
      {
          // 遍历所有待发送的传输消息
          foreach (var unicastOp in transportOps.UnicastTransportOperations)
          {
              // 仅拦截发往第三方目标队列的消息，不影响其他业务消息
              if (unicastOp.Destination.Equals("ThirdParty_Target_Queue"))
              {
                  unicastOp.Properties.SetRabbitMqPropertiesModification(props =>
                  {
                      props.ContentEncoding = "utf-8";
                      // 自定义header场景用：props.Headers["encoding_type"] = "utf-8";
                  });
              }
          }
          // 调用原生派发器完成后续发送流程
          await _innerCoreDispatcher.Dispatch(transportOps, transaction, cancellationToken);
      }
  }
  

验证方式 #

配置完成后，可通过RabbitMQ管理控制台查看目标队列的消息详情，确认encoding_type属性已正确设置为utf-8即可。

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