Mass Transit结合Azure Service Bus死信队列及拓扑控制问题

一、手动控制消息投递至Azure Service Bus死信队列并添加自定义属性 #

要实现自主决定将消息送入死信队列，同时添加自定义错误属性，可按以下步骤操作：

1. 禁用Mass Transit自动重试与内置错误队列
   在配置中关闭自动重试，避免消息因重试耗尽才进入死信队列；同时禁用内置错误队列，确保消息流向完全由你控制：

   services.AddMassTransit(x =>
   {
       x.AddConsumer<YourMessageConsumer>();
   
       x.UsingAzureServiceBus((context, cfg) =>
       {
           cfg.Host("your-azure-service-bus-connection-string");
   
           // 全局禁用自动重试
           cfg.UseMessageRetry(r => r.None());
   
           cfg.ReceiveEndpoint("your-target-queue", e =>
           {
               e.ConfigureConsumer<YourMessageConsumer>(context);
               // 禁止将失败消息发送到Mass Transit内置错误队列
               e.DiscardFaultedMessages();
           });
       });
   });
   

2. 在消费者中手动投递死信并添加自定义属性
   在消息处理逻辑中捕获异常后，通过Azure Service Bus原生上下文将消息投递至死信队列，同时添加自定义错误属性：

   public class YourMessageConsumer : IConsumer<YourMessage>
   {
       private readonly ILogger<YourMessageConsumer> _logger;
   
       public YourMessageConsumer(ILogger<YourMessageConsumer> logger)
       {
           _logger = logger;
       }
   
       public async Task Consume(ConsumeContext<YourMessage> context)
       {
           try
           {
               // 业务处理逻辑
               throw new Exception("模拟可修复错误");
           }
           catch (Exception ex)
           {
               // 记录错误日志
               _logger.LogError(ex, "处理消息 {MessageId} 失败", context.MessageId);
   
               // 获取Azure Service Bus原生接收上下文
               if (context.TryGetPayload<AzureServiceBusReceiveContext>(out var asbContext))
               {
                   // 构建自定义错误属性
                   var errorProperties = new Dictionary<string, object>
                   {
                       ["ErrorCategory"] = "可修复错误",
                       ["ExceptionMessage"] = ex.Message,
                       ["OccurredTime"] = DateTime.UtcNow.ToString("o")
                   };
   
                   // 投递到死信队列，附带错误原因、描述及自定义属性
                   await asbContext.DeadLetterAsync(
                       reason: "业务处理失败",
                       description: ex.Message,
                       properties: errorProperties);
               }
           }
       }
   }
   

   处理后，消息会直接进入Azure Service Bus原生死信队列，自定义属性会被添加到死信消息的ApplicationProperties中。

二、完全禁用Mass Transit自动创建队列/主题/订阅 #

要彻底控制拓扑结构，禁止Mass Transit自动创建任何资源，需从全局和端点两个层面配置：

1. 全局配置禁用拓扑自动创建
   在Azure Service Bus配置中，设置全局跳过资源创建，并禁用消息拓扑的自动注册：

   x.UsingAzureServiceBus((context, cfg) =>
   {
       cfg.Host("your-azure-service-bus-connection-string");
   
       // 全局禁用消息拓扑的自动创建（包括主题、订阅等）
       cfg.Topology.RegisterMessageTopology<YourMessage>(false);
   
       // 配置所有端点时默认跳过创建
       cfg.ConfigureEndpoints(context, endpointConfig =>
       {
           endpointConfig.SkipCreation = true;
       });
   });
   

2. 单个接收端点明确跳过创建
   针对每个接收端点，显式设置SkipCreation = true；如果是订阅主题的端点，也要对订阅配置跳过创建：

   cfg.ReceiveEndpoint("your-queue", e =>
   {
       e.ConfigureConsumer<YourMessageConsumer>(context);
       // 跳过当前队列的自动创建
       e.SkipCreation = true;
   
       // 如果是订阅主题的场景，跳过订阅的自动创建
       e.Subscribe("your-topic", subscribeConfig =>
       {
           subscribeConfig.SkipCreation = true;
       });
   });
   

   注意：配置完成后，必须确保所有需要使用的队列、主题、订阅已通过Azure Portal、Azure CLI或其他方式手动创建完成，否则消费者无法正常连接和接收消息。

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