如何用@RetryableTopic判断Kafka消费者是否重试及实现单次执行逻辑

有两种直接可行的方式：

• 检查消息头：Spring Kafka的@RetryableTopic会自动为重试消息添加kafka_retry_attempt请求头，首次消费时该值为0，每重试一次递增1。只要该值≥1，就说明消息已经至少重试过一次。另外也可以检查retry_topic-original-topic头，存在该头则表示当前消息是从重试/DLT路由过来的。
• 直接注入重试次数：在消费方法的参数列表中通过@Header注解直接获取重试次数，无需手动遍历headers，示例代码：

fun consume(
    record: ConsumerRecord<SomeEventKey, SomeEvent>,
    @Header(name = KafkaHeaders.RETRY_ATTEMPT, required = false) retryAttempt: Int?
) {
    val hasRetried = retryAttempt?.let { it > 0 } ?: false
    // 根据hasRetried判断后续逻辑
}

核心思路是基于重试状态做判断或通过消息唯一标识实现幂等，确保操作只执行一次。以下是两种实用方案：

方案1：基于重试头判断，仅首次处理时执行一次性操作 #

如果你的一次性操作需要在首次消费时执行（不管后续是否触发重试），可以通过重试次数判断，仅当retryAttempt == 0时执行前后操作：

@Service
class KafkaEventListener(private val myServiceEventHandler: MyServiceEventHandler) {

    @RetryableTopic(
        attempts = "10",
        backoff = Backoff(
            delayExpression = "180000",
            maxDelayExpression = "1200000",
            multiplierExpression = "2"
        ),
        topicSuffixingStrategy = TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE,
    )
    @KafkaListener(topics = ["request"])
    fun consume(
        record: ConsumerRecord<SomeEventKey, SomeEvent>,
        @Header(name = KafkaHeaders.RETRY_ATTEMPT, required = false) retryAttempt: Int? = 0
    ) {
        val isFirstAttempt = retryAttempt == 0
        val event = record.value()
        var success = false

        try {
            if (isFirstAttempt) {
                // 仅首次执行的前置操作，比如初始化资源
                println("执行前置一次性操作：${event.id}")
            }

            // 原有业务逻辑
            if (isFirstAttempt) {
                myServiceEventHandler.handleRequestEventByThrowException()
            } else {
                myServiceEventHandler.handleRequestEvent(event)
            }

            success = true
            if (isFirstAttempt) {
                // 仅首次执行的后置操作，比如统计成功次数
                myServiceEventHandler.incrementSuccessCount()
                println("执行后置一次性操作：${event.id}")
            }
        } catch (e: Exception) {
            // 抛出异常触发重试
            throw e
        }
    }

    @DltHandler
    fun handleDltMessage(someEvent: SomeEvent) {
        myServiceEventHandler.handleDltEvent(someEvent)
    }
}

方案2：基于消息唯一标识实现幂等（更可靠） #

如果你的一次性操作需要在消息最终成功时执行一次（不管重试多少次），推荐用消息的唯一ID（比如事件中的id字段）做幂等，记录已执行的操作，避免重复统计：

// 先给MyServiceEventHandler新增幂等相关方法
interface MyServiceEventHandler {
    fun handleRequestEvent(event: SomeEvent)
    fun handleRequestEventByThrowException()
    fun handleDltEvent(event: SomeEvent)
    // 标记操作已启动，返回是否是首次标记
    fun markOperationStarted(eventId: String): Boolean
    // 仅当未统计过时递增成功次数，返回是否执行了统计
    fun incrementSuccessCountIfNotExists(eventId: String): Boolean
}

@Service
class KafkaEventListener(private val myServiceEventHandler: MyServiceEventHandler) {

    @RetryableTopic(
        attempts = "10",
        backoff = Backoff(
            delayExpression = "180000",
            maxDelayExpression = "1200000",
            multiplierExpression = "2"
        ),
        topicSuffixingStrategy = TopicSuffixingStrategy.SUFFIX_WITH_INDEX_VALUE,
    )
    @KafkaListener(topics = ["request"])
    fun consume(record: ConsumerRecord<SomeEventKey, SomeEvent>) {
        val event = record.value()
        val eventId = event.id // 假设事件包含唯一ID字段

        try {
            // 前置操作：仅未执行过时才执行
            if (myServiceEventHandler.markOperationStarted(eventId)) {
                println("执行前置一次性操作：$eventId")
            }

            // 原有业务逻辑
            if (record.headers().any { it.key().contains("retry_topic-attempts") }) {
                myServiceEventHandler.handleRequestEvent(event)
            } else {
                myServiceEventHandler.handleRequestEventByThrowException()
            }

            // 后置统计：仅未统计过时才执行
            if (myServiceEventHandler.incrementSuccessCountIfNotExists(eventId)) {
                println("统计成功次数：$eventId")
            }
        } catch (e: Exception) {
            // 抛出异常触发重试
            throw e
        }
    }

    @DltHandler
    fun handleDltMessage(someEvent: SomeEvent) {
        myServiceEventHandler.handleDltEvent(someEvent)
    }
}

方案对比 #

• 方案1适合操作仅需在首次消费时执行的场景，但如果首次消费失败触发重试，后续重试不会再执行一次性操作。
• 方案2更可靠，通过幂等性保证即使多次重试，成功后的统计操作也只会执行一次，适合需要确保最终一致性的场景。

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