Spring Boot Kafka - 如何处理毒丸(Poison Pill）消息

ゝ一纸荒年。 2022-09-05 06:29 19阅读 0赞

### 文章目录 ###

*  Spring Boot Kafka - 如何处理毒丸(Poison Pill）消息
 *   *  前言
     *  模拟毒丸(Poison Pill）消息
     *  使用ErrorHandlingDeserializer处理反序列化失败
     *  参考文档

# Spring Boot Kafka - 如何处理毒丸(Poison Pill）消息 #

## 前言 ##

在[Spring Boot Kafka - 序列化和反序列化JSON][Spring Boot Kafka - _JSON] 一文中描述了如何在发送消息和接收消息时作JSON序列化和反序列化。

但是上文只考虑了最乐观的情况，如何接收消息时反序列化失败，会怎么样？

如果反序列化失败，就会出现毒丸(Poison Pill）现象，Consumer会卡在“反序列化失败-重试-反序列化失败”的死循环中，无法再处理后续消息。

错误消息类似：

java.lang.IllegalStateException: **This error handler cannot process 'SerializationException's directly; please consider configuring an 'ErrorHandlingDeserializer' in the value and/or key deserializer**
    
    ...
    
    Caused by: org.apache.kafka.common.errors.SerializationException: Error deserializing key/value for partition users-0 at offset 17. If needed, please seek past the record to continue consumption.

## 模拟毒丸(Poison Pill）消息 ##

利用Kafka脚本来发送消息：

kafka-console-producer.sh --bootstrap-server kafka:9092 --topic users
    
    # 输入任意字符串，按回车

这时候发送的消息是未经JSON序列化的消息，所以会导致Consumer的JSON反序列化失败。

## 使用ErrorHandlingDeserializer处理反序列化失败 ##

在`application.yaml` 中配置ErrorHandlingDeserializer反序列化。

server:
      port: 8080
    spring:
      kafka:
        consumer:
          bootstrap-servers: localhost:9092
          group-id: demo-group
          auto-offset-reset: earliest
          key-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
          value-deserializer: org.springframework.kafka.support.serializer.ErrorHandlingDeserializer
          properties:
            spring.json.trusted.packages: '*'
            spring.deserializer.key.delegate.class: org.apache.kafka.common.serialization.StringDeserializer
            spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JsonDeserializer
    
        producer:
          bootstrap-servers: localhost:9092
          key-serializer: org.apache.kafka.common.serialization.StringSerializer
          value-serializer: org.springframework.kafka.support.serializer.JsonSerializer

说明：

*  将Consumer的`key-deserializer` 和 `value-deserializer` 都配置为 `org.springframework.kafka.support.serializer.ErrorHandlingDeserializer`
 *  并委任具体的Key和Value反序列化器：

spring.deserializer.key.delegate.class: org.apache.kafka.common.serialization.StringDeserializer
    spring.deserializer.value.delegate.class: org.springframework.kafka.support.serializer.JsonDeserializer

*  在Key或Value反序列化失败时，`ErrorHandlingDeserializer` 确保毒丸(Poison Pill）消息被处理掉并记录日志，Consumer offeset可以向前移动，使得Consumer可以继续处理后续的消息。

## 参考文档 ##

*  [Spring for Apache Kafka – Beyond the Basics: Can Your Kafka Consumers Handle a Poison Pill?][Spring for Apache Kafka _ Beyond the Basics_ Can Your Kafka Consumers Handle a Poison Pill] 对毒丸(Poison Pill）现象的背后原理和解决方法有详细的讲解。
 *  [Emailing microservice with Apache Kafka and Spring Boot made easy][]
 *  [Spring Kafka - Using ErrorHandlingDeserializer][]
 *  [Better way of error handling in Kafka Consumer][]

[Spring Boot Kafka - _JSON]: https://cookcode.blog.csdn.net/article/details/119762662
[Spring for Apache Kafka _ Beyond the Basics_ Can Your Kafka Consumers Handle a Poison Pill]: https://www.confluent.io/blog/spring-kafka-can-your-kafka-consumers-handle-a-poison-pill/
[Emailing microservice with Apache Kafka and Spring Boot made easy]: https://www.sipios.com/blog-tech/emailing-microservice-apache-kafka-spring-boot
[Spring Kafka - Using ErrorHandlingDeserializer]: https://docs.spring.io/spring-kafka/reference/html/#error-handling-deserializer
[Better way of error handling in Kafka Consumer]: https://stackoverflow.com/questions/63236346/better-way-of-error-handling-in-kafka-consumer