Spring Cloud Stream操作Azure ServiceBus死信队列技术咨询

可以不用Azure原生Java SDK，仅通过Spring Cloud Stream结合Azure Service Bus绑定器实现死信队列的收发、设置死信原因等操作。以下是具体实现方案：

1. 收发死信队列 #

死信队列是原队列的附属队列，Azure Service Bus默认命名格式为[原队列名]$deadletterqueue，Spring Cloud Stream可以通过这个特定名称直接绑定死信队列：

配置文件（application.yml） #

spring:
  cloud:
    stream:
      azure:
        servicebus:
          connection-string: ${AZURE_SERVICEBUS_CONNECTION_STRING}
      bindings:
        # 原队列的输入绑定
        main-input:
          destination: my-business-queue
          group: business-consumer-group
        # 死信队列的输入绑定（接收死信消息）
        dlq-input:
          destination: my-business-queue$deadletterqueue
          group: dlq-consumer-group
        # 死信队列的输出绑定（手动发送死信消息）
        dlq-output:
          destination: my-business-queue$deadletterqueue

死信队列消费者代码 #

@Service
public class DlqMessageHandler {
    @StreamListener("dlq-input")
    public void processDlqMessage(String payload, @Headers Map<String, Object> headers) {
        // 处理死信消息，比如记录日志、重试或归档
        String dlqReason = (String) headers.get(ServiceBusMessageHeaders.DEAD_LETTER_REASON);
        String dlqDesc = (String) headers.get(ServiceBusMessageHeaders.DEAD_LETTER_DESCRIPTION);
        System.out.printf("Received DLQ message: %s, Reason: %s, Description: %s%n", payload, dlqReason, dlqDesc);
    }
}

2. 设置死信原因与描述 #

自动转发失败消息到死信队列 #

当原队列的消息处理失败时，抛出AmqpRejectAndDontRequeueException并通过消息头指定死信原因和描述，绑定器会自动将消息转发到死信队列：

@Service
public class MainQueueMessageHandler {
    @StreamListener("main-input")
    public void processMainMessage(String payload) {
        try {
            // 模拟业务处理失败
            throw new IllegalArgumentException("参数格式错误");
        } catch (Exception e) {
            // 设置死信元数据
            Map<String, Object> dlqHeaders = Map.of(
                ServiceBusMessageHeaders.DEAD_LETTER_REASON, "业务处理失败",
                ServiceBusMessageHeaders.DEAD_LETTER_DESCRIPTION, e.getMessage()
            );
            throw new AmqpRejectAndDontRequeueException("转发到死信队列", new MessageHeaders(dlqHeaders));
        }
    }
}

手动发送消息到死信队列 #

如果需要主动将消息发送到死信队列，直接通过死信队列的输出绑定发送，并设置对应头信息：

@Service
public class DlqMessageSender {
    @Autowired
    @Output("dlq-output")
    private MessageChannel dlqOutputChannel;

    public void sendMessageToDlq(String payload, String reason, String description) {
        Message<?> dlqMessage = MessageBuilder.withPayload(payload)
                .setHeader(ServiceBusMessageHeaders.DEAD_LETTER_REASON, reason)
                .setHeader(ServiceBusMessageHeaders.DEAD_LETTER_DESCRIPTION, description)
                .build();
        dlqOutputChannel.send(dlqMessage);
    }
}

3. 关键注意事项 #

• 必须确保原队列已在Azure端开启死信转发（可通过Azure Portal或CLI配置），否则失败消息不会自动进入死信队列。
• 使用Spring Cloud Stream Azure绑定器版本需≥4.0.0，旧版本对死信队列的支持不完善。
• 死信队列的绑定名称必须严格遵循Azure的默认命名规则：原队列名$deadletterqueue，否则无法正确绑定。

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