From e97bd3f223be53b02531b46f701aa69d65098cc7 Mon Sep 17 00:00:00 2001 From: jun <2395306536@qq.com> Date: Thu, 10 Aug 2023 19:27:23 +0800 Subject: [PATCH] update kafka-questions-01.md --- .../message-queue/kafka-questions-01.md | 98 +++++++++++++++++++ 1 file changed, 98 insertions(+) diff --git a/docs/high-performance/message-queue/kafka-questions-01.md b/docs/high-performance/message-queue/kafka-questions-01.md index 7f124abb..f67eb2ad 100644 --- a/docs/high-performance/message-queue/kafka-questions-01.md +++ b/docs/high-performance/message-queue/kafka-questions-01.md @@ -218,6 +218,104 @@ acks 的默认值即为 1，代表我们的消息被 leader 副本接收之后 - 处理完消息再提交：依旧有消息重复消费的风险，和自动提交一样 - 拉取到消息即提交：会有消息丢失的风险。允许消息延时的场景，一般会采用这种方式。然后，通过定时任务在业务不繁忙（比如凌晨）的时候做数据兜底。 +# kafka的重试机制 + 网上关于spring kafka的默认重试机制文章很多，但大多都是过时的，和实际运行结果完全不一样。以下代码是根据 spring-kafka-2.9.3 源码重新梳理一下。 + +## 消费失败会怎么样 +在消费过程中，当其中一个消息消费异常时，会不会卡住后续队列消息的消费？这样业务岂不是卡住了？ + +生产者代码： + ```Java + for (int i = 0; i < 10; i++) { + kafkaTemplate.send(KafkaConst.TEST_TOPIC, String.valueOf(i)) + } + ``` +消费者消代码： + ```Java + @KafkaListener(topics = {KafkaConst.TEST_TOPIC},groupId = "apple") + private void customer(String message) throws InterruptedException { + log.info("kafka customer:{}",message); + Integer n = Integer.parseInt(message); + if (n%5==0){ + throw new RuntimeException(); + } + } + ``` + +在默认配置下，当消费异常会进行重试，重试多次后会跳过当前消息，继续进行后续消息的消费，不会一直卡在当前消息。下面是一段消费的日志，可以看出当 test-0@95 重试多次后会被跳过。 +```Java +2023-08-10 12:03:32.918 DEBUG 9700 --- [ntainer#0-0-C-1] o.s.kafka.listener.DefaultErrorHandler : Skipping seek of: test-0@95 +2023-08-10 12:03:32.918 TRACE 9700 --- [ntainer#0-0-C-1] o.s.kafka.listener.DefaultErrorHandler : Seeking: test-0 to: 96 +2023-08-10 12:03:32.918 INFO 9700 --- [ntainer#0-0-C-1] o.a.k.clients.consumer.KafkaConsumer : [Consumer clientId=consumer-apple-1, groupId=apple] Seeking to offset 96 for partition test-0 + +``` + +## 默认会重试多少次？ +默认配置下，消费异常会进行重试，重试次数是多少, 重试是否有时间间隔？ +10次。看源码 FailedRecordTracker 类有个 recovered 函数，返回 Boolean 值判断是否要进行重试，下面是这个函数中判断是否重试的逻辑： +```Java + FailedRecord failedRecord = getFailedRecordInstance(record, exception, map, topicPartition); + this.retryListeners.forEach(rl -> + rl.failedDelivery(record, exception, failedRecord.getDeliveryAttempts().get())); + long nextBackOff = failedRecord.getBackOffExecution().nextBackOff(); + if (nextBackOff != BackOffExecution.STOP) { + this.backOffHandler.onNextBackOff(container, exception, nextBackOff); + return false; + } +``` +其中 BackOffExecution.STOP 的值为-1，nextBackOff 的值调用 BackOff 类的 nextBackOff() 函数。如果当前执行次数大于最大执行次数则返回 STOP，既超过这个最大执行次数后才会停止重试。 +``` +public long nextBackOff() { + this.currentAttempts++; + if (this.currentAttempts <= getMaxAttempts()) { + return getInterval(); + } + else { + return STOP; + } + } +``` +那么这个 getMaxAttempts 的值又是多少呢？回到最开始，当执行出错会进入 DefaultErrorHandler 。DefaultErrorHandler 默认的构造函数是： +```Java + public DefaultErrorHandler() { + this(null, SeekUtils.DEFAULT_BACK_OFF); + } +``` +SeekUtils.DEFAULT_BACK_OFF 定义的是 +```Java +public static final int DEFAULT_MAX_FAILURES = 10; + +public static final FixedBackOff DEFAULT_BACK_OFF = new FixedBackOff(0, DEFAULT_MAX_FAILURES - 1); +``` +DEFAULT_MAX_FAILURES 的值是10，currentAttempts从0到9，所以总共会执行10次，每次重试的时间间隔为0。 + +## 重试失败后的数据如何再次处理 + +当达到最大重试次数后，数据会直接被跳过，继续向后进行。当代码修复后，如何重新消费这些重试失败的数据呢？ +死信队列（Dead Letter Queue，简称DLQ）是消息中间件中的一种特殊队列。它主要用于处理无法被消费者正确处理的消息，通常是因为消息格式错误、处理失败、消费超时等情况导致的消息被"丢弃"或"死亡"的情况。 +当消息进入队列后，消费者会尝试处理它。如果处理失败，或者超过一定的重试次数仍无法被成功处理，消息可以发送到死信队列中，而不是被永久性地丢弃。在死信队列中，可以进一步分析、处理这些无法正常消费的消息，以便定位问题、修复错误，并采取适当的措施。 + +spring kafka 中只需要加上 `@DltHandler` 注解即可将重试失败的消息推到死信队列，死信队列的topic是在原 topic 后加上 '.DLT'。然后开启新的消费者消费死信队列即可。 +```Java + @DltHandler + @KafkaListener(topics = {KafkaConst.TEST_TOPIC}, groupId = "apple") + private void customer(String message) { + log.info("kafka customer:{}", message); + Integer n = Integer.parseInt(message); + if (n % 5 == 0) { + throw new RuntimeException(); + } + System.out.println(n); + } + + @KafkaListener(topics = {KafkaConst.TEST_TOPIC + ".DLT"}, groupId = "apple") + private void delCustomer(String message) { + // + } +``` +## 如何自定义重试次数，以及时间间隔 +"......，未完待续。" + ### Reference - Kafka 官方文档：https://kafka.apache.org/documentation/