Spring Cloud Stream集成Azure Service Bus重试策略配置失效问题

问题场景 #

使用Spring Cloud Stream集成Azure Service Bus，版本与依赖如下：

<spring-cloud-azure.version>4.3.0</spring-cloud-azure.version>
<spring-cloud.version>2021.0.3</spring-cloud.version>

<dependency>
    <groupId>com.azure.spring</groupId>
    <artifactId>spring-cloud-azure-starter</artifactId>
</dependency>

<dependency>
    <groupId>org.springframework.cloud</groupId>
    <artifactId>spring-cloud-stream</artifactId>
</dependency>

<dependency>
    <groupId>com.azure.spring</groupId>
    <artifactId>spring-cloud-azure-stream-binder-servicebus</artifactId>
    <version>4.3.0</version>
</dependency>

期望实现业务异常触发后的延迟重试策略，但配置后未生效：抛出异常时消息被无延迟重试10次，实际是被放弃后重新从队列接收，而非执行配置的重试逻辑，同时日志重复出现Dispatcher has no subscribers for channel 'application.errorChannel'错误。

原application.yml配置：

spring:
  cloud:
    stream:
      bindings:
        onReceive-in-0:
          destination: subscriber-1-input-queue
    azure:
      servicebus:
        consumer:
          retry:
            exponential:
              base-delay: 100
              max-delay: 1000
              max-retries: 5
            fixed:
              delay: 100
              max-retries: 5
            mode: exponential
        connection-string: Endpoint=...

消费者代码：

@Slf4j
@Component
public class ValuesConsumer {
    @Bean
    public Consumer<String> onReceive() {
        return (message) -> {
          log.info("Received the value {} in Consumer", message);
            switch (message) {
                case "IntegrationException" -> throw new IntegrationException();
                case "PoisonPillException" -> throw new PoisonPillException();
                case "ProcessingException" -> throw new ProcessingException();
                case "ValidationException" -> throw new ValidationException();
            }
        };
    }
}

问题原因 #

1. 配置层级错误：原配置将重试规则放在spring.azure.servicebus.consumer.retry下，这是Azure Service Bus客户端的底层重试配置（针对网络故障等场景），而非Spring Cloud Stream消费端的业务异常重试配置。
2. 双重重试冲突：Azure Service Bus默认会对放回队列的消息执行10次无延迟重试，覆盖了自定义配置。
3. 异常未处理：消费端抛出异常后，Spring Integration将异常发送到errorChannel但无订阅者，导致消息被强制重新投递。

正确解决方案 #

1. 调整重试配置到消费端绑定属性 #

使用Spring Cloud Stream标准的重试配置，针对具体绑定设置延迟重试规则，同时禁用Azure Service Bus客户端的底层重试，避免冲突：

spring:
  cloud:
    stream:
      bindings:
        onReceive-in-0:
          destination: subscriber-1-input-queue
          consumer:
            # 总尝试次数（含第一次消费），对应5次重试需设为6
            max-attempts: 6
            back-off:
              # 初始延迟（毫秒）
              initial-interval: 100
              # 最大延迟（毫秒）
              max-interval: 1000
              # 指数退避乘数，每次延迟乘以该值
              multiplier: 2
            # 开启死信队列，重试失败后消息转入死信队列，避免无限循环
            enable-dlq: true
    azure:
      servicebus:
        connection-string: Endpoint=...
        consumer:
          # 禁用Azure Service Bus客户端自身重试，避免双重重试
          retry:
            max-retries: 0

2. 处理ErrorChannel异常（可选） #

添加errorChannel订阅者，避免日志报错并自定义异常处理逻辑：

@Slf4j
@Component
public class ErrorMessageHandler {
    @ServiceActivator(inputChannel = "errorChannel")
    public void handleError(ErrorMessage errorMessage) {
        Throwable exception = errorMessage.getPayload();
        log.error("消费异常，消息内容: {}, 异常信息: {}", 
                  errorMessage.getOriginalMessage(), exception.getMessage(), exception);
        // 可扩展：记录异常到监控系统、触发告警等
    }
}

配置说明 #

• max-attempts：控制总消费尝试次数，max-attempts = 重试次数 + 1（含第一次消费）。
• back-off：配置指数退避规则，initial-interval为首次重试延迟，multiplier控制延迟增长倍数，max-interval限制最大延迟时间。
• enable-dlq：开启后，重试失败的消息会自动转入对应队列的死信队列，可后续人工排查处理。
• spring.azure.servicebus.consumer.retry.max-retries: 0：禁用Azure Service Bus客户端的底层重试，确保仅执行自定义的业务重试逻辑。

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