SpringBoot消息重试机制怎么配置？自动重试保证数据一致性的方法

<dependency>
    <groupId>org.springframework.retry</groupId>
    <artifactId>spring-retry</artifactId>
</dependency>
<dependency>
    <groupId>org.springframework</groupId>
    <artifactId>spring-aspects</artifactId>
</dependency>

@SpringBootApplication
@EnableRetry  // 启用重试功能
public class Application {
    public static void main(String[] args) {
        SpringApplication.run(Application.class, args);
    }
}

@Service
public class OrderService {
    /**
     * 处理订单方法
     * 当抛出 RuntimeException 时自动重试
     */
    @Retryable(
        retryFor = RuntimeException.class,    // 对哪些异常重试
        maxAttempts = 3,                      // 最大重试次数
        backoff = @Backoff(                   // 退避策略
            delay = 1000,                     // 初始延迟时间（毫秒）
            multiplier = 2,                   // 延迟倍数
            maxDelay = 10000                  // 最大延迟时间
        )
    )
    public void processOrder(String orderId) {
        log.info("处理订单: {}, 时间: {}", orderId, LocalDateTime.now());
        // 模拟业务异常
        if (new Random().nextBoolean()) {
            throw new RuntimeException("网络超时");
        }
        log.info("订单处理成功: {}", orderId);
    }
    /**
     * 重试失败后的回调方法
     */
    @Recover
    public void recover(RuntimeException e, String orderId) {
        log.error("订单处理最终失败: {}, 原因: {}", orderId, e.getMessage());
        // 可以记录到数据库或发送告警
    }
}

@Service
public class RetryService {
    /**
     * 手动重试方法
     */
    public <T> T executeWithRetry(Callable<T> task, int maxAttempts, long delayMs) {
        int attempt = 0;
        Exception lastException = null;
        while (attempt < maxAttempts) {
            try {
                attempt++;
                log.info("第 {} 次尝试", attempt);
                return task.call();
            } catch (Exception e) {
                lastException = e;
                log.warn("第 {} 次尝试失败: {}", attempt, e.getMessage());
                if (attempt < maxAttempts) {
                    try {
                        Thread.sleep(delayMs * attempt); // 指数退避
                    } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt();
                        break;
                    }
                }
            }
        }
        throw new RuntimeException("重试 " + maxAttempts + " 次后仍然失败", lastException);
    }
}
// 使用示例
@Service
public class OrderService {
    private final RetryService retryService;
    public OrderService(RetryService retryService) {
        this.retryService = retryService;
    }
    public void processOrder(String orderId) {
        retryService.executeWithRetry(() -> {
            // 业务逻辑
            processOrderInternal(orderId);
            return null;
        }, 3, 1000);
    }
}

spring:
  kafka:
    consumer:
      enable-auto-commit: false           # 手动提交偏移量
      auto-offset-reset: earliest         # 失败后从最早开始消费
    listener:
      ack-mode: manual                    # 手动确认模式
      retry:
        max-attempts: 3                   # 最大重试次数
        initial-interval: 1000            # 初始重试间隔（毫秒）
        multiplier: 2                     # 间隔倍数
        max-interval: 10000               # 最大间隔时间

@Configuration
public class KafkaConfig {
    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, OrderMessage> kafkaListenerContainerFactory(
        ConsumerFactory<String, OrderMessage> consumerFactory,
        KafkaTemplate<String, OrderMessage> kafkaTemplate) {
        ConcurrentKafkaListenerContainerFactory<String, OrderMessage> factory = 
            new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory);
        // 创建死信队列恢复器
        DeadLetterPublishingRecoverer recoverer = new DeadLetterPublishingRecoverer(kafkaTemplate);
        // 设置错误处理器：重试3次后发送到死信队列
        SeekToCurrentErrorHandler errorHandler = new SeekToCurrentErrorHandler(
            recoverer, 
            new FixedBackOff(1000, 3)  // 每次重试间隔1秒，最多重试3次
        );
        factory.setErrorHandler(errorHandler);
        return factory;
    }
}

@Component
public class CustomRetryPolicy implements RetryPolicy {
    private static final int MAX_ATTEMPTS = 3;
    private static final long MIN_DELAY = 1000;
    private static final long MAX_DELAY = 10000;
    @Override
    public boolean canRetry(RetryContext context) {
        return context.getRetryCount() < MAX_ATTEMPTS;
    }
    @Override
    public RetryContext open(RetryContext parent) {
        return new DefaultRetryContext(parent);
    }
    @Override
    public void close(RetryContext context) {
        // 清理资源
    }
}

@Configuration
public class CircuitBreakerConfig {
    @Bean
    public CircuitBreaker circuitBreaker() {
        return CircuitBreaker.builder()
            .failureThreshold(50)        // 失败率超过50%触发熔断
            .waitDurationInOpenState(Duration.ofSeconds(30))  // 熔断30秒
            .build();
    }
}

@Retryable(
    retryFor = {NetworkException.class, TimeoutException.class},  // 可重试异常
    exclude = {IllegalArgumentException.class}                   // 不可重试异常
)
public void processOrder(String orderId) {
    // 业务逻辑
}

@Retryable(
    retryFor = RuntimeException.class,
    maxAttempts = 3
)
public void processOrder(String orderId) {
    log.info("开始处理订单: {}", orderId);
    // 业务逻辑
    log.info("订单处理成功: {}", orderId);
}
@Recover
public void recover(RuntimeException e, String orderId) {
    log.error("订单处理最终失败，已重试3次: {}, 原因: {}", orderId, e.getMessage());
    // 记录到数据库
    retryLogRepository.save(new RetryLog(orderId, e.getMessage()));
    // 发送告警通知
    alertService.sendAlert("订单处理失败", orderId);
}

@Service
public class OrderRetryService {
    private static final int MAX_RETRY = 3;
    private static final long INITIAL_DELAY = 1000;
    private final RestTemplate restTemplate;
    private final OrderRepository orderRepository;
    public OrderRetryService(RestTemplate restTemplate, OrderRepository orderRepository) {
        this.restTemplate = restTemplate;
        this.orderRepository = orderRepository;
    }
    /**
     * 通知库存服务扣减库存
     */
    public void notifyInventory(String orderId) {
        String url = "http://inventory-service/api/inventory/deduct";
        retryWithExponentialBackoff(() -> {
            ResponseEntity<String> response = restTemplate.postForEntity(
                url, 
                new InventoryRequest(orderId), 
                String.class
            );
            if (!response.getStatusCode().is2xxSuccessful()) {
                throw new RuntimeException("库存服务返回失败: " + response.getStatusCode());
            }
            log.info("库存扣减成功: {}", orderId);
        }, MAX_RETRY, INITIAL_DELAY);
    }
    /**
     * 指数退避重试方法
     */
    private void retryWithExponentialBackoff(Runnable task, int maxAttempts, long initialDelay) {
        int attempts = 0;
        long delay = initialDelay;
        while (attempts < maxAttempts) {
            try {
                attempts++;
                task.run();
                return; // 成功则返回
            } catch (Exception e) {
                log.warn("第 {} 次尝试失败: {}", attempts, e.getMessage());
                if (attempts < maxAttempts) {
                    try {
                        log.info("等待 {} 毫秒后重试", delay);
                        Thread.sleep(delay);
                        delay *= 2; // 指数增长
                    } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt();
                        throw new RuntimeException("重试被中断", ie);
                    }
                }
            }
        }
        // 所有重试都失败
        log.error("已重试 {} 次，任务仍然失败", maxAttempts);
        throw new RuntimeException("任务重试失败");
    }
}