kafka-重试和死信主题(SpringBoot整合Kafka)

文章目录

  • 1、重试和死信主题
  • 2、死信队列
  • 3、代码演示
    • 3.1、appication.yml
    • 3.2、引入spring-kafka依赖
    • 3.3、创建SpringBoot启动类
    • 3.4、创建生产者发送消息
    • 3.5、创建消费者消费消息

1、重试和死信主题

kafka默认支持重试和死信主题

  1. 重试主题:当消费者消费消息异常时,手动ack如果有异常继续向后消费,为了保证消息尽可能成功消费,可以将消费异常的消息扔到重试主题,再通过重试主题的消费者尝试消费
  2. 死信主题:当一个消息消费多次仍然失败,可以将该消息存到死信主题。避免漏消息,以后可以人工介入手动解决消息消费失败的问题

2、死信队列

在Kafka中,DLT通常指的是 Dead Letter Topic(死信队列)

Dead Letter Topic(DLT)的定义与功能

  1. 背景:原生Kafka是不支持Retry Topic和DLT的,但Spring Kafka在客户端实现了这两个功能。
  2. 功能:当消息在Kafka中被消费时,如果消费逻辑抛出异常,并且重试策略(如默认重试10次后)仍无法成功处理该消息,Spring Kafka会将该消息发送到DLT中。
  3. 自定义处理:可以通过自定义SeekToCurrentErrorHandler来控制消费失败后的处理逻辑,例如添加重试间隔、设置重试次数等。如果在重试后消息仍然消费失败,Spring Kafka会将该消息发送到DLT。

DLT的使用与意义

  1. 错误处理:DLT为Kafka提供了一种机制来处理那些无法被成功消费的消息,从而避免了消息的丢失或阻塞。
  2. 监控与告警:通过对DLT的监控,可以及时发现并处理那些无法被成功消费的消息,从而保障Kafka系统的稳定性和可靠性。
  3. 二次处理:对于发送到DLT的消息,可以进行二次处理,如手动干预、修复数据等,以确保这些消息能够最终得到处理。

总之,在Kafka中,DLT是一个用于处理无法被成功消费的消息的特殊Topic,它提供了一种灵活且可靠的机制来保障Kafka系统的稳定性和可靠性。

3、代码演示

3.1、appication.yml

server:port: 8120
# v1
spring:Kafka:bootstrap-servers: 192.168.74.148:9095,192.168.74.148:9096,192.168.74.148:9097consumer:# read-committed读事务已提交的消息 解决脏读问题isolation-level: read-committed # 消费者的事务隔离级别:read-uncommitted会导致脏读,可以读取生产者事务还未提交的消息# 消费者是否自动ack :true自动ack 消费者获取到消息后kafka提交消费者偏移量# 调用ack方法时才会提交ack给kafka
#      enable-auto-commit: false# 消费者提交ack时多长时间批量提交一次auto-commit-interval: 1000# 消费者第一次消费主题消息时从哪个位置开始# earliest:从最早的消息开始消费# latest:第一次从LEO位置开始消费# none:如果主题分区没有偏移量,则抛出异常auto-offset-reset: earliest  #指定Offset消费:earliest | latest | nonekey-deserializer: org.apache.kafka.common.serialization.StringDeserializervalue-deserializer: org.apache.kafka.common.serialization.StringDeserializer# json反序列化器
#      value-deserializer: org.springframework.kafka.support.serializer.JsonDeserializerproperties:spring.json.trusted.packages: "*"listener:# 手动ack:manual手动ack时 如果有异常会尝试一直消费
#      ack-mode: manual# 手动ack:消费有异常时停止ack-mode: manual_immediate

3.2、引入spring-kafka依赖

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><parent><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-parent</artifactId><version>3.0.5</version><relativePath/> <!-- lookup parent from repository --></parent><!-- Generated by https://start.springboot.io --><!-- 优质的 spring/boot/data/security/cloud 框架中文文档尽在 => https://springdoc.cn --><groupId>com.atguigu</groupId><artifactId>spring-kafka-consumer</artifactId><version>0.0.1-SNAPSHOT</version><name>spring-kafka-consumer</name><description>spring-kafka-consumer</description><properties><java.version>17</java.version></properties><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter</artifactId></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-test</artifactId><scope>test</scope></dependency><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId></dependency></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build></project>

3.3、创建SpringBoot启动类

package com.atguigu.spring.kafka.consumer;import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;// Generated by https://start.springboot.io
// 优质的 spring/boot/data/security/cloud 框架中文文档尽在 => https://springdoc.cn
@SpringBootApplication
public class SpringKafkaConsumerApplication {public static void main(String[] args) {SpringApplication.run(SpringKafkaConsumerApplication.class, args);}}

3.4、创建生产者发送消息

package com.atguigu.spring.kafka.consumer;
import jakarta.annotation.Resource;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.kafka.core.KafkaTemplate;
import java.time.LocalDateTime;
@SpringBootTest
public class KafkaProducerApplicationTests {@ResourceKafkaTemplate kafkaTemplate;@Testvoid sendRertyMsg(){kafkaTemplate.send("retry_topic",0,"","重试机制和死信队列"+"_"+ LocalDateTime.now());}
}

在这里插入图片描述

没有指定分区,默认会创建一个分区,即使此时有3个kafka实例,也只会用一个,因为只有一个分区

在这里插入图片描述
在这里插入图片描述

[[{"partition": 0,"offset": 0,"msg": "重试机制和死信队列_2024-06-07T15:49:37.398841600","timespan": 1717746578357,"date": "2024-06-07 07:49:38"}]
]

3.5、创建消费者消费消息

给一个消费者配重试主题的时候,死信主题消费者一般不写

package com.atguigu.spring.kafka.consumer.listener;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.RetryableTopic;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.retry.annotation.Backoff;
@Component
public class MyKafkaListenerAck {/*** 自动ack可能会导致漏消息*      spring-kafka:*                     自动ack 如果有异常,会死循环获取消息重新消费*                        不能继续向后消费消息,会导致消息积压**                    手动ack 配置了手动ack,且ack-mode为manual_immediate时,*                        如果消息消费失败,会继续向后消费* @param record*///指定消费者消费异常使用重试+死信主题   必须结合手动ack使用@RetryableTopic(attempts = "3",numPartitions = "3",  //重试主题的分区数量backoff =@Backoff(value = 2_000L),  //重试的时间间隔autoCreateTopics ="true",retryTopicSuffix = "-myRetry",  //指定重试主题创建时的后缀名:使用原主题的名称拼接后缀生成名称 -retrydltTopicSuffix = "-myDlt" //指定DLT主题创建时的后缀名:使用原主题的名称拼接后缀生成名称 -dlt)@KafkaListener(groupId = "my_group1",topicPartitions = {@TopicPartition(topic = "retry_topic",partitions = {"0","1","2"})})public void consumeByRetry(ConsumerRecord<String, String> record, Acknowledgment ack) {System.out.println("consumeByRetry消费者获取到消息:topic = "+ record.topic()+",partition:"+record.partition()+",offset = "+record.offset()+",key = "+record.key()+",value = "+record.value());int i = 1/0;//手动ackack.acknowledge();}/* //死信队列默认名称在原队列后拼接'-dlt'@KafkaListener(groupId = "my_group2",topicPartitions = {@TopicPartition(topic = "topic_dlt",partitions = {"0","1","2"})})public void consumeByDLT(ConsumerRecord<String, String> record, Acknowledgment ack) {System.out.println("consumeByDLT消费者获取到消息:topic = "+ record.topic()+",partition:"+record.partition()+",offset = "+record.offset()+",key = "+record.key()+",value = "+record.value());//手动ackack.acknowledge();}*/}

此时运行SpringKafkaConsumerApplication,控制台会报错,因为我们手动写了异常 int i=1/0

在这里插入图片描述

在这里插入图片描述

在这里插入图片描述

[[{"partition": 0,"offset": 0,"msg": "重试机制和死信队列_2024-06-07T15:49:37.398841600","timespan": 1717746979414,"date": "2024-06-07 07:56:19"}]
]

我们发现原本在重试主题中的消息,死信主题也有一份

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处:http://www.rhkb.cn/news/343792.html

如若内容造成侵权/违法违规/事实不符,请联系长河编程网进行投诉反馈email:809451989@qq.com,一经查实,立即删除!

相关文章

android中调用onnxruntime框架

创建空白项目 安装Android Studio及创建空白项目参考&#xff1a;【安卓Java原生开发学习记录】一、安卓开发环境的搭建与HelloWorld&#xff08;详细图文解释&#xff09;_安卓原生开发-CSDN博客 切记&#xff1a;build configuration language 一定选择Groovy&#xff01;官…

Java——IO流(一)-(1/8):File、IO流概述、File文件对象的创建(介绍、实例演示)

目录 File IO流概述 File文件对象的创建 介绍 实例演示 File 存储数据的方案 变量 double money 9999.5 数组 int[] age new int[100];对象 Student s new Student()集合 List<Student> students new ArrayList<>()…

Chrome 源码阅读:跟踪一个鼠标事件的流程

我们通过在关键节点打断点的方式&#xff0c;去分析一个鼠标事件的流程。 我们知道chromium是多进程模型&#xff0c;那么&#xff0c;我们可以推测&#xff1a;一个鼠标消息先从主进程产生&#xff0c;再通过跨进程通信发送给渲染进程&#xff0c;渲染进程再发送给WebFrame&a…

L45---506.相对名次(java)--排序

1.题目描述 2.知识点 &#xff08;1&#xff09;String.join(" ", words) 是 Java 中的一个语法&#xff0c;用于将数组或集合中的元素连接成一个单独的字符串&#xff0c;连接时使用指定的分隔符。这里的 " " 是作为分隔符使用的一个空格字符串。 Strin…

4、后端本地环境搭建

后端本地环境搭建 4.1 安装jdk 下载完成后双击安装的 jdk &#xff0c;点下一步&#xff0c;选择安装目录&#xff0c;一直点下一步&#xff0c;直到结束。 安装完成后同样需要配置环境变量 window s 搜索查看高级系统设置—— 高级 —— 环境变量 —— 系统变量 1、新建一…

API接口测试工具:jmeter的安装、汉化、Jmeter桌面快捷图标和基本使用

文章目录 测试工具&#xff1a;JmeterJmeter安装和配置Jmeter汉化设置中文语言&#xff1a;永久方式设置中文语言&#xff1a;临时方式 设置Jmeter桌面快捷图标jmeter基本用法Jmeter无法保存测试问题解决 测试工具&#xff1a;Jmeter Jmeter依赖于JDK&#xff0c;所以必须确保…

第十一届蓝桥杯C++青少年组中/高级组国赛2020年10月真题解析

一、单选题 第1题 在数组中&#xff0c;数组名表示&#xff08; &#xff09;. A:数组第1个元素的首地址 B:数组第2个元素的首地址 C:数组所有元素的首地址&#xff0c; D:数组最后1个元素的首地址 答案&#xff1a;A 数组名是一个地址&#xff0c;指向第一个元素 第2题 …

中继器、集线器、网桥、交换机、路由器和网关

目录 前言一、中继器、集线器1.1 中继器1.2 集线器 二、网桥、交换机2.1 网桥2.1.1 认识网桥2.1.2 网桥的工作原理2.1.3 生成树网桥 2.2 交换机2.2.1 交换机的特征2.2.2 交换机的交换模式2.2.3 交换机的功能 三、路由器、网关3.1 路由器的介绍3.2 路由器的工作过程3.2.1 前置知…

多客圈子论坛系统 httpGet 任意文件读取漏洞复现

0x01 产品简介 多客圈子论坛系统是一种面向特定人群或特定话题的社交网络&#xff0c;它提供了用户之间交流、分享、讨论的平台。在这个系统中&#xff0c;用户可以创建、加入不同的圈子&#xff0c;圈子可以是基于兴趣、地域、职业等不同主题的。用户可以在圈子中发帖、评论、…

17- Redis 中的 quicklist 数据结构

在 Redis 3.0 之前&#xff0c;List 对象的底层数据结构是双向链表或者压缩列表&#xff0c;然后在 Redis 3.2 的时候&#xff0c;List 对象的底层改由 quicklist 数据结构实现。 其实 quicklist 就是【双向链表 压缩列表】组合&#xff0c;因为一个 quicklist 就是一个链表&…

【产品研发】NPDP价值作用概述

导读&#xff1a;本文结合个人实践和思考对NPDP的价值和作用做了概述说明&#xff0c;对于产品经理而言掌握NPDP的知识体系并且应用到实际工作中&#xff0c;这是非常有必要的。走出以往狭隘的产品研发工作认知&#xff0c;以开放心态学习国际化产品创新开发流程将极大提升产品…

深度学习的舌象诊断:从舌头上了解系统性疾病!

首先 深度学习算法能否解决东方医学中依靠医生经验的诊断问题&#xff1f;而要实现这个目标&#xff0c;需要什么呢&#xff1f; 用舌头诊断被称为口腔健康的指标&#xff0c;但在东方医学中&#xff0c;舌头也被用来评估全身的状况。换句话说&#xff0c;通过分析舌头的图像…

Java学习-JDBC(一)

JDBC 概念 JDBC(Java Database Connectivity)Java数据库连接JDBC提供了一组独立于任何数据库管理系统的APIJava提供接口规范&#xff0c;由各个数据库厂商提供接口的实现&#xff0c;厂商提供的实现类封装成jar文件&#xff0c;也就是我们俗称的数据库驱动jar包JDBC充分体现了…

C#上位机开发

目录 一、上位机简介二、C#语法三、新建VS工程四、WinForm控件4.1 属性4.2 事件4.3 窗体方法4.4 常用控件4.5 布局 五、Serial上位机5.1 UI界面设计5.2 功能设计 六、项目打包成安装包6.1 前提准备6.2 打包步骤 一、上位机简介 在单片机项目开发中&#xff0c;上位机也是一个很…

SwiftUI六组合复杂用户界面

代码下载 应用的首页是一个纵向滚动的地标类别列表&#xff0c;每一个类别内部是一个横向滑动列表。随后将构建应用的页面导航&#xff0c;这个过程中可以学习到如果组合各种视图&#xff0c;并让它们适配不同的设备尺寸和设备方向。 下载起步项目并跟着本篇教程一步步实践&a…

【C语言训练题库】扫雷->简单小游戏!

&#x1f525;博客主页&#x1f525;&#xff1a;【 坊钰_CSDN博客 】 欢迎各位点赞&#x1f44d;评论✍收藏⭐ 目录 1. 题目 2. 解析 3. 代码 4. 小结 1. 题目 小sun上课的时候非常喜欢玩扫雷。他现小sun有一个初始的雷矩阵&#xff0c;他希望你帮他生成一个扫雷矩阵。 扫雷…

Vue3_对接腾讯云COS_大文件分片上传和下载

目录 一、腾讯云后台配置 二、安装SDK 1.script 引入方式 2.webpack 引入方式 三、文件上传 1.new COS 实例 2.上传文件 四、文件下载 腾讯云官方文档&#xff1a; 腾讯云官方文档https://cloud.tencent.com/document/product/436/11459 一、腾讯云后台配置 1.登录 对…

Codeforces Round 951 (Div. 2) F. Kostyanych‘s Theorem(思维题 交互好题)

题目 交互题&#xff0c;n&#xff08;n<1e5&#xff09;个点的完全图&#xff0c;无向的&#xff0c;初始恰好删了n-2条边 每次询问可以输入一个d&#xff1a;? d 交互器会输出一个当前度>d的点v&#xff0c; 如果有多个这样的点&#xff0c;输出度最小的&#xff…

【CS.CN】优化HTTP传输:揭示Transfer-Encoding: chunked的奥秘与应用

文章目录 0 序言0.1 由来0.2 使用场景 1 Transfer-Encoding: chunked的机制2 语法 && 通过设置Transfer-Encoding: chunked优化性能3 总结References 0 序言 0.1 由来 Transfer-Encoding头部字段在HTTP/1.1中被引入&#xff0c;用于指示数据传输过程中使用的编码方式…

InternLM-XComposer2-4KHD开拓性的4K高清视觉-语言模型

大型视觉-语言模型&#xff08;LVLM&#xff09;在图像字幕和视觉问答&#xff08;VQA&#xff09;等任务中表现出色。然而&#xff0c;受限于分辨率&#xff0c;这些模型在处理包含细微视觉内容的图像时面临挑战。 分辨率的限制严重阻碍了模型处理含有丰富细节的图像的能力。…