springboot 整合 kafka 錯誤重試機制

在 Spring Boot 中整合 Kafka 時，可以通過配置 RetryTemplate 和 KafkaListenerEndpointRegistrar 來實現錯誤重試機制。以下是一個簡單的示例：

1. 首先，在 pom.xml 文件中添加 Kafka 和 Spring Retry 的依賴：

<dependencies>
    <!-- Kafka -->
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    <!-- Spring Retry -->
    <dependency>
        <groupId>org.springframework.retry</groupId>
        <artifactId>spring-retry</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework</groupId>
        <artifactId>spring-aspects</artifactId>
    </dependency>
</dependencies>

2. 在 application.yml 或 application.properties 文件中配置 Kafka 和 Retry 的相關屬性：

spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: my-group
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer
  retry:
    enabled: true
    max-attempts: 3
    backoff:
      initial-interval: 1000
      multiplier: 2
      max-interval: 10000

3. 創建一個配置類，用于配置 RetryTemplate 和 KafkaListenerEndpointRegistrar：

@Configuration
public class KafkaRetryConfig {

    @Bean
    public RetryTemplate retryTemplate() {
        SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
        retryPolicy.setMaxAttempts(3);

        ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
        backOffPolicy.setInitialInterval(1000L);
        backOffPolicy.setMultiplier(2);
        backOffPolicy.setMaxInterval(10000L);

        RetryTemplate retryTemplate = new RetryTemplate();
        retryTemplate.setRetryPolicy(retryPolicy);
        retryTemplate.setBackOffPolicy(backOffPolicy);

        return retryTemplate;
    }

    @Bean
    public KafkaListenerEndpointRegistrar kafkaListenerEndpointRegistrar(RetryTemplate retryTemplate) {
        KafkaListenerEndpointRegistrar registrar = new KafkaListenerEndpointRegistrar();
        registrar.setRetryTemplate(retryTemplate);
        return registrar;
    }
}

4. 創建一個消費者監聽器，并使用 @KafkaListener 注解標注。在這個例子中，我們創建一個簡單的 ConsumerRecord 處理類：

@Service
public class KafkaConsumerListener {

    @KafkaListener(topics = "my-topic", groupId = "my-group")
    public void listen(ConsumerRecord<String, String> record) {
        System.out.printf("Received message: key = %s, value = %s, partition = %d, offset = %d%n",
                record.key(), record.value(), record.partition(), record.offset());
    }
}

現在，當消費者接收到消息時，如果處理過程中發生錯誤，Spring Retry 將自動重試，直到達到最大重試次數。