Spring Cloud集成RabbitMQ：实现非阻塞指数退避重试机制

核心思路 #

基于Reactor响应式生态的retryWhen操作符实现带次数跟踪的重试逻辑，结合StreamBridge动态路由消息：遇到确定性错误直接发送至死信队列（DLQ）；非确定性错误重试耗尽后，根据累计失败次数发送至对应超时队列，完全适配Spring Cloud Stream响应式消费者模型。

具体代码实现 #

1. 注入StreamBridge #

用于动态发送消息到指定队列：

@Autowired
private StreamBridge streamBridge;

2. 定义错误分类方法 #

明确两类错误的判断逻辑：

// 判断是否为非确定性错误（可重试，如网络超时、服务临时不可用）
private boolean isRepeatableError(Throwable throwable) {
    return throwable instanceof ConnectTimeoutException || throwable instanceof ResourceAccessException;
}

// 判断是否为确定性错误（不可重试，如数据格式错误、业务校验失败）
private boolean isDeterministicError(Throwable throwable) {
    return throwable instanceof InvalidFormatException || throwable instanceof BusinessValidationException;
}

3. 重构消费者Bean #

整合重试、错误路由逻辑：

@Bean
public Function<Flux<Message<byte[]>>, Mono<Void>> myReactiveConsumer() {
    return messageFlux -> messageFlux
            .flatMap(this::processWithRetryAndRouting)
            .then();
}

private Mono<Void> processWithRetryAndRouting(Message<byte[]> message) {
    // 定义重试策略：最多重试3次，指数退避间隔，仅针对非确定性错误
    Retry retryStrategy = Retry.backoff(3, Duration.ofSeconds(1))
            .filter(this::isRepeatableError)
            .doBeforeRetry(retrySignal -> {
                // 将重试次数存入消息头，用于后续路由
                int retryCount = retrySignal.totalRetries();
                message.getHeaders().put("x-retry-count", retryCount);
            });

    return Mono.fromRunnable(() -> processMessage(message))
            .retryWhen(retryStrategy)
            .onErrorResume(throwable -> {
                if (isDeterministicError(throwable)) {
                    // 确定性错误：发送至DLQ
                    streamBridge.send("dlq-out-0", message);
                } else {
                    // 非确定性错误重试耗尽：根据次数路由到对应超时队列
                    int totalFailures = ((Integer) message.getHeaders().getOrDefault("x-retry-count", 0)) + 1;
                    String targetQueue = String.format("timeout-queue-%d-out-0", totalFailures);
                    streamBridge.send(targetQueue, message);
                }
                return Mono.empty();
            });
}

// 原业务处理方法
private void processMessage(Message<byte[]> message) {
    // 业务逻辑实现，可能抛出异常
}

4. 队列绑定配置 #

在application.yml中配置输出绑定（根据你的消息中间件调整）：

spring:
  cloud:
    stream:
      bindings:
        dlq-out-0:
          destination: dlq-topic
          binder: kafka
        timeout-queue-1-out-0:
          destination: timeout-topic-1
        timeout-queue-2-out-0:
          destination: timeout-topic-2
        timeout-queue-3-out-0:
          destination: timeout-topic-3

关键说明 #

• 抛弃阻塞式RetryTemplate，使用Reactor原生Retry组件，完美适配响应式消费者，直接获取重试次数上下文。
• 通过StreamBridge动态指定输出目标，无需提前定义大量@OutputBean，灵活匹配不同重试次数的超时队列。
• 用filter隔离错误类型，确保仅非确定性错误进入重试流程，避免无效重试消耗资源。

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