Apache Kafka-消费端消费重试和死信队列
文章目录
- 概述
- Code
- POM依赖
- 配置文件
- 配置类
- SeekToCurrentErrorHandler
- 自定义逻辑处理消费异常
- 生产者
- 消费者
- 单元测试
- 测速结果
- 源码地址
概述
Spring-Kafka 提供消费重试的机制。当消息消费失败的时候,Spring-Kafka 会通过消费重试机制,重新投递该消息给 Consumer ,让 Consumer 重新消费消息 。
默认情况下,Spring-Kafka 达到配置的重试次数时,【每条消息的失败重试时间,由配置的时间隔决定】Consumer 如果依然消费失败 ,那么该消息就会进入到死信队列。
Spring-Kafka 封装了消费重试和死信队列, 将正常情况下无法被消费的消息称为死信消息(Dead-Letter Message
),将存储死信消息的特殊队列称为死信队列(Dead-Letter Queue
)。
我们在应用中可以对死信队列中的消息进行监控重发,来使得消费者实例再次进行消费,消费端需要做幂等性的处理。
Code
POM依赖
<dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><!-- 引入 Spring-Kafka 依赖 --><dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>junit</groupId><artifactId>junit</artifactId><scope>test</scope></dependency></dependencies>
配置文件
spring:# Kafka 配置项,对应 KafkaProperties 配置类kafka:bootstrap-servers: 192.168.126.140:9092 # 指定 Kafka Broker 地址,可以设置多个,以逗号分隔# Kafka Producer 配置项producer:acks: 1 # 0-不应答。1-leader 应答。all-所有 leader 和 follower 应答。retries: 3 # 发送失败时,重试发送的次数key-serializer: org.apache.kafka.common.serialization.StringSerializer # 消息的 key 的序列化value-serializer: org.springframework.kafka.support.serializer.JsonSerializer # 消息的 value 的序列化# Kafka Consumer 配置项consumer:auto-offset-reset: earliest # 设置消费者分组最初的消费进度为 earliestkey-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.springframework.kafka.support.serializer.JsonDeserializerproperties:spring:json:trusted:packages: com.artisan.springkafka.domain# Kafka Consumer Listener 监听器配置listener:missing-topics-fatal: false # 消费监听接口监听的主题不存在时,默认会报错。所以通过设置为 false ,解决报错logging:level:org:springframework:kafka: ERROR # spring-kafkaapache:kafka: ERROR # kafka
配置类
首先要写一个配置类,用于处理消费异常 ErrorHandler
package com.artisan.springkafka.configuration;import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.*;
import org.springframework.util.backoff.BackOff;
import org.springframework.util.backoff.FixedBackOff;/*** @author 小工匠* @version 1.0* @description: TODO* @date 2021/2/18 14:32* @mark: show me the code , change the world*/@Configuration
public class KafkaConfiguration {private Logger logger = LoggerFactory.getLogger(getClass());@Bean@Primarypublic ErrorHandler kafkaErrorHandler(KafkaTemplate<?, ?> template) {logger.warn("kafkaErrorHandler begin to Handle");// <1> 创建 DeadLetterPublishingRecoverer 对象ConsumerRecordRecoverer recoverer = new DeadLetterPublishingRecoverer(template);// <2> 创建 FixedBackOff 对象 设置重试间隔 10秒 次数为 3次BackOff backOff = new FixedBackOff(10 * 1000L, 3L);// <3> 创建 SeekToCurrentErrorHandler 对象return new SeekToCurrentErrorHandler(recoverer, backOff);}// @Bean
// @Primary
// public BatchErrorHandler kafkaBatchErrorHandler() {// // 创建 SeekToCurrentBatchErrorHandler 对象
// SeekToCurrentBatchErrorHandler batchErrorHandler = new SeekToCurrentBatchErrorHandler();
// // 创建 FixedBackOff 对象
// BackOff backOff = new FixedBackOff(10 * 1000L, 3L);
// batchErrorHandler.setBackOff(backOff);
// // 返回
// return batchErrorHandler;
// }
}
Spring-Kafka 通过实现自定义的 SeekToCurrentErrorHandler ,当 Consumer 消费消息异常的时候,进行拦截处理:
- 重试小于最大次数时,重新投递该消息给 Consumer
- 重试到达最大次数时,如果Consumer 还是消费失败时,该消息就会发送到死信队列。 死信队列的 命名规则为: 原有 Topic + .DLT 后缀 = 其死信队列的 Topic
ConsumerRecordRecoverer recoverer = new DeadLetterPublishingRecoverer(template);
创建 DeadLetterPublishingRecoverer 对象,它负责实现,在重试到达最大次数时,Consumer 还是消费失败时,该消息就会发送到死信队列。
BackOff backOff = new FixedBackOff(10 * 1000L, 3L);
也可以选择 BackOff 的另一个子类 ExponentialBackOff 实现,提供指数递增的间隔时间
new SeekToCurrentErrorHandler(recoverer, backOff);
创建 SeekToCurrentErrorHandler 对象,负责处理异常,串联整个消费重试的整个过程。
SeekToCurrentErrorHandler
在消息消费失败时,SeekToCurrentErrorHandler
会将 调用 Kafka Consumer 的 seek(TopicPartition partition, long offset)
方法,将 Consumer 对于该消息对应的 TopicPartition 分区的本地进度设置成该消息的位置。
这样,Consumer 在下次从 Kafka Broker 拉取消息的时候,又能重新拉取到这条消费失败的消息,并且是第一条。
同时,Spring-Kafka 使用 FailedRecordTracker 对每个 Topic 的每个 TopicPartition 消费失败次数进行计数,这样相当于对该 TopicPartition 的第一条消费失败的消息的消费失败次数进行计数。
另外,在 FailedRecordTracker
中,会调用 BackOff 来进行计算,该消息的下一次重新消费的时间,通过 Thread#sleep(...)
方法,实现重新消费的时间间隔。
注意:
FailedRecordTracker 提供的计数是客户端级别的,重启 JVM 应用后,计数是会丢失的。所以,如果想要计数进行持久化,需要自己重新实现下 FailedRecordTracker 类,通过 ZooKeeper 存储计数。
SeekToCurrentErrorHandler 是只针对消息的单条消费失败的消费重试处理。如果想要有消息的批量消费失败的消费重试处理,可以使用 SeekToCurrentBatchErrorHandler 。配置方式如下
@Bean
@Primary
public BatchErrorHandler kafkaBatchErrorHandler() {// 创建 SeekToCurrentBatchErrorHandler 对象SeekToCurrentBatchErrorHandler batchErrorHandler = new SeekToCurrentBatchErrorHandler();// 创建 FixedBackOff 对象BackOff backOff = new FixedBackOff(10 * 1000L, 3L);batchErrorHandler.setBackOff(backOff);// 返回return batchErrorHandler;
}
SeekToCurrentBatchErrorHandler
暂时不支持死信队列的机制。
自定义逻辑处理消费异常
支持自定义 ErrorHandler 或 BatchErrorHandler 实现类,实现对消费异常的自定义的逻辑
比如 https://github.com/spring-projects/spring-kafka/blob/master/spring-kafka/src/main/java/org/springframework/kafka/listener/LoggingErrorHandler.java
public class LoggingErrorHandler implements ErrorHandler {private static final LogAccessor LOGGER = new LogAccessor(LogFactory.getLog(LoggingErrorHandler.class));@Overridepublic void handle(Exception thrownException, ConsumerRecord<?, ?> record) {LOGGER.error(thrownException, () -> "Error while processing: " + ObjectUtils.nullSafeToString(record));}}
配置方式同 SeekToCurrentErrorHandler
或 SeekToCurrentBatchErrorHandler
。
生产者
package com.artisan.springkafka.producer;import com.artisan.springkafka.constants.TOPIC;
import com.artisan.springkafka.domain.MessageMock;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;import java.util.Random;
import java.util.concurrent.ExecutionException;/*** @author 小工匠* @version 1.0* @description: TODO* @date 2021/2/17 22:25* @mark: show me the code , change the world*/@Component
public class ArtisanProducerMock {@Autowiredprivate KafkaTemplate<Object,Object> kafkaTemplate ;public ListenableFuture<SendResult<Object, Object>> sendMsgASync() {// 模拟发送的消息Integer id = new Random().nextInt(100);MessageMock messageMock = new MessageMock(id,"messageSendByAsync-" + id);// 异步发送消息ListenableFuture<SendResult<Object, Object>> result = kafkaTemplate.send(TOPIC.TOPIC, messageMock);return result ;}}
消费者
package com.artisan.springkafka.consumer;import com.artisan.springkafka.domain.MessageMock;
import com.artisan.springkafka.constants.TOPIC;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;/*** @author 小工匠* @version 1.0* @description: TODO* @date 2021/2/17 22:33* @mark: show me the code , change the world*/@Component
public class ArtisanCosumerMock {private Logger logger = LoggerFactory.getLogger(getClass());private static final String CONSUMER_GROUP_PREFIX = "MOCK-A" ;@KafkaListener(topics = TOPIC.TOPIC ,groupId = CONSUMER_GROUP_PREFIX + TOPIC.TOPIC)public void onMessage(MessageMock messageMock){logger.info("【接受到消息][线程:{} 消息内容:{}]", Thread.currentThread().getName(), messageMock);// 模拟抛出一次一行throw new RuntimeException("MOCK Handle Exception Happened");}}
在消费消息时候,抛出一个 RuntimeException 异常,模拟消费失败
单元测试
package com.artisan.springkafka.produceTest;import com.artisan.springkafka.SpringkafkaApplication;
import com.artisan.springkafka.producer.ArtisanProducerMock;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.support.SendResult;
import org.springframework.test.context.junit4.SpringRunner;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;/*** @author 小工匠* * @version 1.0* @description: TODO* @date 2021/2/17 22:40* @mark: show me the code , change the world*/@RunWith(SpringRunner.class)
@SpringBootTest(classes = SpringkafkaApplication.class)
public class ProduceMockTest {private Logger logger = LoggerFactory.getLogger(getClass());@Autowiredprivate ArtisanProducerMock artisanProducerMock;@Testpublic void testAsynSend() throws ExecutionException, InterruptedException {logger.info("开始发送");artisanProducerMock.sendMsgASync().addCallback(new ListenableFutureCallback<SendResult<Object, Object>>() {@Overridepublic void onFailure(Throwable throwable) {logger.info(" 发送异常{}]]", throwable);}@Overridepublic void onSuccess(SendResult<Object, Object> objectObjectSendResult) {logger.info("回调结果 Result = topic:[{}] , partition:[{}], offset:[{}]",objectObjectSendResult.getRecordMetadata().topic(),objectObjectSendResult.getRecordMetadata().partition(),objectObjectSendResult.getRecordMetadata().offset());}});// 阻塞等待,保证消费new CountDownLatch(1).await();}}
测速结果
我们把这个日志来梳理一下
2021-02-18 16:18:08.032 INFO 25940 --- [ main] c.a.s.produceTest.ProduceMockTest : 开始发送
2021-02-18 16:18:08.332 INFO 25940 --- [ad | producer-1] c.a.s.produceTest.ProduceMockTest : 回调结果 Result = topic:[C_RT_TOPIC] , partition:[0], offset:[0]
2021-02-18 16:18:08.371 INFO 25940 --- [ntainer#0-0-C-1] c.a.s.consumer.ArtisanCosumerMock : 【接受到消息][线程:org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 消息内容:MessageMock{id=15, name='messageSendByAsync-15'}]
2021-02-18 16:18:18.384 ERROR 25940 --- [ntainer#0-0-C-1] essageListenerContainer$ListenerConsumer : Error handler threw an exception......
......
......2021-02-18 16:18:18.388 INFO 25940 --- [ntainer#0-0-C-1] c.a.s.consumer.ArtisanCosumerMock : 【接受到消息][线程:org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 消息内容:MessageMock{id=15, name='messageSendByAsync-15'}]
2021-02-18 16:18:28.390 ERROR 25940 --- [ntainer#0-0-C-1] essageListenerContainer$ListenerConsumer : Error handler threw an exception......
......
......2021-02-18 16:18:28.394 INFO 25940 --- [ntainer#0-0-C-1] c.a.s.consumer.ArtisanCosumerMock : 【接受到消息][线程:org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 消息内容:MessageMock{id=15, name='messageSendByAsync-15'}]
2021-02-18 16:18:38.395 ERROR 25940 --- [ntainer#0-0-C-1] essageListenerContainer$ListenerConsumer : Error handler threw an exception......
......
......2021-02-18 16:18:38.399 INFO 25940 --- [ntainer#0-0-C-1] c.a.s.consumer.ArtisanCosumerMock : 【接受到消息][线程:org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1 消息内容:MessageMock{id=15, name='messageSendByAsync-15'}]
清晰了么 老兄?
是不是和我们设置的消费重试
BackOff backOff = new FixedBackOff(10 * 1000L, 3L);
10秒 重试3次
3次处理后依然失败,转入死信队列
看看数据
源码地址
https://github.com/yangshangwei/boot2/tree/master/springkafkaRetries
Apache Kafka-消费端消费重试和死信队列相关推荐
- RabbitMQ消费端消费机制
一.确认机制 第一种 public static void getMessage() throws Exception {Connection connection = ConnectionUtil. ...
- 6.RokcketMQ消息重试与死信队列
消息发送重试机制 Producer对发送失败的消息进行重新发送的机制,称为消息发送重试机制,也称为消息重投机制. 对于消息重投,需要注意以下几点: 生产者在发送消息时,若采用同步或异步发送方式,发送失 ...
- RabbitMQ重试机制+死信队列
RabbitMQ的基本使用.ACK确认机制这里就不赘述了,这里主要是想实现一个应用场景: 消息消费失败后重试至多三次,仍失败则加入死信队列 一.重试机制 首先说一下RabbitMQ的消息重试机制,顾名 ...
- 面试官:说说RabbitMQ 消费端限流、TTL、死信队列
欢迎关注方志朋的博客,回复"666"获面试宝典 1. 为什么要对消费端限流 假设一个场景,首先,我们 Rabbitmq 服务器积压了有上万条未处理的消息,我们随便打开一个消费者客户 ...
- kafka怎么保证数据消费一次且仅消费一次?使用消息队列如何保证幂等性?
精确一次处理语义(exactly onece semantic–EOS),Kafka的EOS主要体现在3个方面: 1)幂等producer 保证单个分区的只会发送一次,不会出现重复消息 2)事务(tr ...
- kafka指定分区消费
前言 在某些场景中,消费者组中不同的消费者需要从某个topic下的不同分区接收消息,并确保消息的有序性,或者producer端需要根据一定的业务规则对消息进行分类然后发送到不同的分区,然后不同的消费者 ...
- RabbitMQ(八):SpringBoot 整合 RabbitMQ(三种消息确认机制以及消费端限流)
说明 本文 SpringBoot 与 RabbitMQ 进行整合的时候,包含了三种消息的确认模式,如果查询详细的确认模式设置,请阅读:RabbitMQ的三种消息确认模式 同时消费端也采取了限流的措施, ...
- java如何保证mq一定被消费_消费端如何保证消息队列MQ的有序消费
消息无序产生的原因 消息队列,既然是队列就能保证消息在进入队列,以及出队列的时候保证消息的有序性,显然这是在消息的生产端(Producer),但是往往在生产环境中有多个消息的消费端(Consumer) ...
- 消息中间件--RabbitMQ ---高级特性之消费端ACK与重回队列
什么是消费端的ACK和重回队列? 消费端的手工ACK和NACK 消费端进行消费的时候,如果由于业务异常我们可以进行日志的记录,然后进行补偿 如果由于服务器宕机等严重问题,那我们就需要手工进行ACK保障 ...
最新文章
- Android 使用 ADB 命令录制屏幕上的视频
- 获取当前元素在兄弟元素节点中的索引
- HDU 6108.小C的倍数问题 (2017百度之星程序设计大赛 - 初赛(A)1001)
- Leetcode 59. 螺旋矩阵 II (每日一题 20210926)
- mysql 组复制和传统复制_MySQL的GTID复制与传统复制的相互切换
- 【面试招聘】有哪些好的秋招经验分享?数据、算法岗的几点经验分享
- cuda编程_CUDA刷新器:CUDA编程模型
- 【英语学习】【WOTD】purview 释义/词源/示例
- 中国“鸭王”失宠的第8年,终于撑不住了!
- asp.net GridView 时间格式化 设置 短日期格式 去掉0!
- 五步法”判断自媒体创作好与坏的标准
- 全球及中国家具市场消费规模与运营能力状况分析报告2022版
- Oracle分析函数用法
- 联想拯救者y7000p 固态移动硬盘安装Linuxmint19.1 操作系统
- h5算命php源码,H5付费算命PHP源码那么火_付费算命源码如何下载
- 手机app测试方法(一)基本流程
- ajax读取txt出现乱码问题,ajax读取txt文本时乱码的解决方案
- 2022西藏最新消防设施操作员模拟考试试题题库及答案
- Linux命令+shell脚本大全:操作文件系统
- 文件IO——标准IO和文件IO的区别
热门文章
- vue 点击弹出文字_vue事件点击穿透解决大法,看这篇文章就够了
- 相对布局(RelativeLayout)
- OnnxRuntime 性能调优
- sql between
- 整合rpc远程调用_远程过程调用(RPC)
- lwip协议栈优化_干货分享 | KNI性能优化实践
- pytorch 笔记: DQN(experience replay)
- Java实战应用50篇(二)-SSM框架中的设计模式:动态代理
- Linux疑难杂症解决方案100篇(三)-嵌套循环及随机数知多少
- BloomFilter——大规模数据处理利器