Spring Kafka：解决顺序处理与发送场景下的消息丢失问题

针对链式主题消息发送结合@RetryableTopic重试的场景问题，以下是几种实用解决方案，按推荐优先级排序：

1. 带超时的同步等待发送（最小改动，保留@RetryableTopic全功能） #

无需修改ACK模式，仅在发送消息时添加短超时等待，既保证能捕获发送异常触发重试，又将阻塞影响控制在合理范围内。

核心思路：调用KafkaTemplate.send()返回的Future对象的get(timeout, unit)方法，设置短超时（如500ms），捕获发送异常后抛出，让@RetryableTopic自动触发重试逻辑。

代码示例：

@KafkaListener(topics = "topicA")
@RetryableTopic(attempts = "5", backoff = @Backoff(delay = 1000))
public void handleTopicA(String message) throws Exception {
    // 执行业务处理逻辑
    processBusinessLogic(message);
    
    // 发送到下一个主题，带超时等待
    try {
        kafkaTemplate.send("topicB", message).get(500, TimeUnit.MILLISECONDS);
    } catch (InterruptedException | ExecutionException | TimeoutException e) {
        // 解包实际异常并抛出，触发@RetryableTopic重试
        Throwable rootCause = e.getCause() != null ? e.getCause() : e;
        throw new RuntimeException("发送消息到topicB失败", rootCause);
    }
}

优缺点：

• 优点：代码改动极小，完全保留@RetryableTopic的指数退避、死信队列等功能；超时时间设置合理时，仅在发送异常时才会阻塞，对正常性能影响可忽略。
• 缺点：极端场景下（如Kafka集群临时不可用）会短暂阻塞监听器线程。

2. 异步发送回调+手动转发到重试主题（无阻塞，需保证幂等） #

如果完全不想阻塞线程，可以在发送失败的回调中，将原消息转发到当前主题的重试队列，复用@RetryableTopic的现有逻辑处理重试，同时手动ACK原消息避免重复消费。

注意：此方案要求业务处理逻辑必须是幂等的，因为消息会被重新处理一次。

代码示例：

@KafkaListener(topics = "topicA")
@RetryableTopic(attempts = "5", backoff = @Backoff(delay = 1000))
public void handleTopicA(String message, Acknowledgment ack) {
    // 执行幂等性业务处理（比如基于消息ID去重）
    processIdempotentBusinessLogic(message);
    
    kafkaTemplate.send("topicB", message)
        .addCallback(
            sendResult -> {
                // 发送成功，确认原消息
                ack.acknowledge();
            },
            throwable -> {
                // 发送失败，将消息转发到当前主题的重试队列
                String retryTopic = "topicA-retry-0"; // 可通过RetryTopicConfiguration动态获取，避免硬编码
                kafkaTemplate.send(retryTopic, message);
                // 确认原消息，防止原主题重复投递
                ack.acknowledge();
            }
        );
}

优缺点：

• 优点：完全无阻塞，不影响生产者线程性能；复用@RetryableTopic的重试机制。
• 缺点：需要保证业务逻辑幂等；需手动处理重试主题的命名（可通过RetryTopicConfigurationSupport的API动态获取重试主题名称，避免硬编码）。

3. 基于Kafka事务的一致性处理（强一致性，性能略降） #

通过开启Kafka事务，将消息处理和发送操作绑定到同一事务中：发送失败时事务回滚，原消息不会被ACK，@RetryableTopic会捕获异常并将消息转到重试队列。

配置示例： #

# 开启生产者事务
spring.kafka.producer.transaction-id-prefix=kafka-tx-
# 关闭消费者自动提交
spring.kafka.consumer.enable-auto-commit=false
# 监听器使用事务模式
spring.kafka.listener.ack-mode=RECORD
spring.kafka.listener.transactional=true

代码示例： #

@KafkaListener(topics = "topicA")
@RetryableTopic(attempts = "5", backoff = @Backoff(delay = 1000))
@Transactional
public void handleTopicA(String message) throws Exception {
    processBusinessLogic(message);
    // 发送操作纳入事务，失败则事务回滚
    kafkaTemplate.send("topicB", message);
}

优缺点：

• 优点：无需手动处理ACK和回调，事务保证消息处理与发送的一致性；完全保留@RetryableTopic功能。
• 缺点：Kafka事务会带来一定性能开销，适合对数据一致性要求极高的场景。

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