update kafka-questions-01.md

Author: jacketfu0

docs/high-performance/message-queue/kafka-questions-01.md

@@ -219,18 +219,22 @@ acks 的默认值即为 1，代表我们的消息被 leader 副本接收之后
   - 拉取到消息即提交：会有消息丢失的风险。允许消息延时的场景，一般会采用这种方式。然后，通过定时任务在业务不繁忙（比如凌晨）的时候做数据兜底。
 
 # kafka的重试机制
- 网上关于spring kafka的默认重试机制文章很多，但大多都是过时的，和实际运行结果完全不一样。以下是根据 spring-kafka-2.9.3 源码重新梳理一下。
+网上关于 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 {
@@ -243,6 +247,7 @@ acks 的默认值即为 1，代表我们的消息被 leader 副本接收之后
  ```
 
 在默认配置下，当消费异常会进行重试，重试多次后会跳过当前消息，继续进行后续消息的消费，不会一直卡在当前消息。下面是一段消费的日志，可以看出当 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
@@ -251,8 +256,10 @@ acks 的默认值即为 1，代表我们的消息被 leader 副本接收之后
 ```
 
 ## 默认会重试多少次？
+
 默认配置下，消费异常会进行重试，重试次数是多少, 重试是否有时间间隔？
-10次。看源码 FailedRecordTracker 类有个 recovered 函数，返回 Boolean 值判断是否要进行重试，下面是这个函数中判断是否重试的逻辑：
+10 次。看源码 FailedRecordTracker 类有个 recovered 函数，返回 Boolean 值判断是否要进行重试，下面是这个函数中判断是否重试的逻辑：
+
 ```Java
   FailedRecord failedRecord = getFailedRecordInstance(record, exception, map, topicPartition);
   this.retryListeners.forEach(rl ->
@@ -263,7 +270,9 @@ acks 的默认值即为 1，代表我们的消息被 leader 副本接收之后
 			return false;
   }
 ```
-其中 BackOffExecution.STOP 的值为-1，nextBackOff 的值调用 BackOff 类的 nextBackOff() 函数。如果当前执行次数大于最大执行次数则返回 STOP，既超过这个最大执行次数后才会停止重试。
+
+其中 BackOffExecution.STOP 的值为 -1，nextBackOff 的值调用 BackOff 类的 nextBackOff() 函数。如果当前执行次数大于最大执行次数则返回 STOP，既超过这个最大执行次数后才会停止重试。
+
 ```Java
 public long nextBackOff() {
   this.currentAttempts++;
@@ -275,29 +284,76 @@ public long nextBackOff() {
   }
 }
 ```
+
 那么这个 getMaxAttempts 的值又是多少呢？回到最开始，当执行出错会进入 DefaultErrorHandler 。DefaultErrorHandler 默认的构造函数是：
+
 ```Java
 public DefaultErrorHandler() {
   this(null, SeekUtils.DEFAULT_BACK_OFF);
 }
 ```
-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。
 
+## 如何自定义重试次数，以及时间间隔
+
+从上面的代码可以知道，默认错误处理器的重试次数以及时间间隔是由 FixedBackOff 控制的，FixedBackOff 是 DefaultErrorHandler 初始化时默认的。所以自定义重试次数以及时间间隔，只需要在 DefaultErrorHandler 初始化的时候传入自定义的 FixedBackOff 即可。重新实现一个 KafkaListenerContainerFactory ，调用 setCommonErrorHandler 设置新的自定义的错误处理器就可以实现。
+
+```Java
+@Bean
+public KafkaListenerContainerFactory kafkaListenerContainerFactory(ConsumerFactory<String, String> consumerFactory) {
+    ConcurrentKafkaListenerContainerFactory factory = new ConcurrentKafkaListenerContainerFactory();
+    // 自定义重试时间间隔以及次数
+    FixedBackOff fixedBackOff = new FixedBackOff(1000, 5);
+    factory.setCommonErrorHandler(new DefaultErrorHandler(fixedBackOff));
+    factory.setConsumerFactory(consumerFactory);
+    return factory;
+}
+```
+
+## 如何在重试失败后进行告警
+
+自定义重试失败后逻辑，需要手动实现，以下是一个简单的例子，重写 DefaultErrorHandler 的 handleRemaining 函数，加上自定义的告警等操作。
+
+```Java
+@Slf4j
+public class DelErrorHandler extends DefaultErrorHandler {
+
+    public DelErrorHandler(FixedBackOff backOff) {
+        super(null,backOff);
+    }
+
+    @Override
+    public void handleRemaining(Exception thrownException, List<ConsumerRecord<?, ?>> records, Consumer<?, ?> consumer, MessageListenerContainer container) {
+        super.handleRemaining(thrownException, records, consumer, container);
+        log.info("重试多次失败");
+        // 自定义操作
+    }
+}
+```
+DefaultErrorHandler 只是默认的一个错误处理器，Spring kafka 还提供了 CommonErrorHandler 接口。手动实现 CommonErrorHandler 就可以实现更多的自定义操作，有很高的灵活性。例如根据不同的错误类型，实现不同的重试逻辑以及业务逻辑等。
+
 ## 重试失败后的数据如何再次处理
 
 当达到最大重试次数后，数据会直接被跳过，继续向后进行。当代码修复后，如何重新消费这些重试失败的数据呢？
-死信队列（Dead Letter Queue，简称DLQ）是消息中间件中的一种特殊队列。它主要用于处理无法被消费者正确处理的消息，通常是因为消息格式错误、处理失败、消费超时等情况导致的消息被"丢弃"或"死亡"的情况。
-当消息进入队列后，消费者会尝试处理它。如果处理失败，或者超过一定的重试次数仍无法被成功处理，消息可以发送到死信队列中，而不是被永久性地丢弃。在死信队列中，可以进一步分析、处理这些无法正常消费的消息，以便定位问题、修复错误，并采取适当的措施。
 
-spring kafka 中只需要加上 `@DltHandler` 注解即可将重试失败的消息推到死信队列，死信队列的topic是在原 topic 后加上 '.DLT'。然后开启新的消费者消费死信队列即可。
+死信队列（Dead Letter Queue，简称DLQ）是消息中间件中的一种特殊队列。它主要用于处理无法被消费者正确处理的消息，通常是因为消息格式错误、处理失败、消费超时等情况导致的消息被"丢弃"或"死亡"的情况。当消息进入队列后，消费者会尝试处理它。如果处理失败，或者超过一定的重试次数仍无法被成功处理，消息可以发送到死信队列中，而不是被永久性地丢弃。在死信队列中，可以进一步分析、处理这些无法正常消费的消息，以便定位问题、修复错误，并采取适当的措施。
+
+`@RetryableTopic` 是 Spring Kafka 中的一个注解,它用于配置某个 Topic 支持消息重试，更推荐使用这个注解来完成重试。
+
 ```Java
-@DltHandler
+@RetryableTopic(
+        attempts = "5",
+        backoff = @Backoff(delay = 100, maxDelay = 1000)
+)
 @KafkaListener(topics = {KafkaConst.TEST_TOPIC}, groupId = "apple")
 private void customer(String message) {
     log.info("kafka customer:{}", message);
@@ -306,15 +362,18 @@ private void customer(String message) {
         throw new RuntimeException();
     }
     System.out.println(n);
-}   
-
-@KafkaListener(topics = {KafkaConst.TEST_TOPIC + ".DLT"}, groupId = "apple")
-private void delCustomer(String message) {
-  //
 }
 ```
-## 如何自定义重试次数，以及时间间隔
-"......，未完待续。"
+
+这个例子在listen方法上使用@RetryableTopic注解,配置了:
+- 重试5次
+- 重试间隔100毫秒,最大间隔1秒
+
+重试完毕后,如果仍然失败,则会进入消息队列,此时就存在两个队列：
+- test 原队列
+- test-dlt 原队列对应的死信队列
+
+对于死信队列的处理，既可以用 `@DltHandler` 处理，也可以使用 `@KafkaListener` 重新消费。
 
 ### Reference