Spring-Kafka @KafkaListener和errorHandler设置监听异常处理

2021年07月26日08:08:42 科技 1409

Apache Kafka是一个消息队列产品，基于Topic partitions的设计，能达到非常高的消息发送处理性能。Spring创建了一个项目Spring-kafka，封装了Apache Kafka-client，用于在Spring项目里快速集成kafka。除了简单的收发消息外，Spring-kafka还提供了很多高级功能。

一、简单集成

首先需要引入依赖：

<dependency>

 <groupId>org.springframework.kafka</groupId>

 <artifactId>spring-kafka</artifactId>

 <version>2.2.6.RELEASE</version>

</dependency>

然后，添加配置：

spring.kafka.producer.bootstrap-servers=127.0.0.1:9092

二、@KafkaListener的使用

关于@KafkaListener接收消息的能力就不赘述了，但是@KafkaListener的功能不止如此，其他的比较常见的，使用场景比较多的功能点如下：

显式的指定消费哪些Topic和分区的消息；
设置每个Topic以及分区初始化的偏移量；
设置消费线程并发度；
设置消息异常处理器；

使用errorHandler做异常处理，首先实现KafkaListenerErrorHandler接口，然后做一些异常处理，比如记录日志等等。而且注解里的配置，是你自定义实现实例在Spring上下文中的Name。

@Component

public class MyKafkaListenerErrorHandler implements KafkaListenerErrorHandler {

 private static final Logger LOGGER = LoggerFactory.getLogger(MyKafkaListenerErrorHandler.class);

 @Override

 public Object handleError(Message<?> message, ListenerExecutionFailedException exception) {

 LOGGER.error("消息：" + message.toString() + "\n" + "异常：" + exception.getMessage());

 return null;

 }

}

例如，调用的时候，填写beanName，例如errorHandler=“kafkaDefaultListenerErrorHandler”，具体示例代码如下所示：

@Service

public class CargoBookedEventHandler {

private static final Logger LOGGER = LoggerFactory.getLogger(CargoBookedEventHandler.class);

@KafkaListener(topics = DomainEventConstant.CARGO_BOOKED_EVENT_TOPIC, errorHandler = "myKafkaListenerErrorHandler")

public void receiveEvent(ConsumerRecord<String, Object> consumerRecord) {

CargoBookedEvent routedEvent = (CargoBookedEvent)consumerRecord.value();

LOGGER.info("Recevied key='{}', message='{}'", consumerRecord.key(), consumerRecord.value());

 }

}

下面是有意在Topic中发送的错误消息，不符合消费方consumer解析的JSON格式，如图所示：

下面是errorHandler针对错误消息的处理，输出的异常日志：

2021-07-19 11:08:45.991 ERROR 29308 --- [ntainer#0-1-C-1] c.r.t.e.MyKafkaListenerErrorHandler : 消息：GenericMessage [payload=hello world, headers={kafka_offset=4, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@e0bcdd6, kafka_timestampType=CREATE_TIME, type=[B@475a94e, kafka_receivedMessageKey=null, kafka_receivedPartitionId=1, kafka_receivedTopic=CARGO_BOOKED_EVENT_TOPIC, kafka_receivedTimestamp=1626664125969}]

异常：Listener method 'public void com.rickie.tracking.eventhandler.CargoBookedEventHandler.receiveEvent(org.apache.kafka.clients.consumer.ConsumerRecord<java.lang.String, java.lang.Object>)' threw exception; nested exception is java.lang.ClassCastException: java.lang.String cannot be cast to com.rickie.dto.event.CargoBookedEvent

2021-07-19 11:08:50.872 ERROR 29308 --- [ntainer#0-0-C-1] c.r.t.e.MyKafkaListenerErrorHandler : 消息：GenericMessage [payload=hello world, headers={kafka_offset=9, kafka_consumer=org.apache.kafka.clients.consumer.KafkaConsumer@67ecd40c, kafka_timestampType=CREATE_TIME, type=[B@4d795117, kafka_receivedMessageKey=null, kafka_receivedPartitionId=0, kafka_receivedTopic=CARGO_BOOKED_EVENT_TOPIC, kafka_receivedTimestamp=1626664130870}]