kafka-消费者-消费异常处理（SpringBoot整合Kafka）

server/2024/10/18 10:16:36/

文章目录

  • 1、消费异常处理
    • 1.1、application.yml配置
    • 1.2、注册异常处理器
    • 1.3、消费者使用异常处理器
    • 1.4、创建生产者发送消息
    • 1.5、创建SpringBoot启动类
    • 1.6、屏蔽 kafka debug 日志 logback.xml
    • 1.7、引入spring-kafka依赖
    • 1.8、消费者控制台：
      • 1.8.1、第一次启动SpringKafkaConsumerApplication
      • 1.8.n、第n次启动SpringKafkaConsumerApplication

1、消费异常处理

1.1、application.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.StringDeserializerlistener:# 手动ack：manual手动ack时 如果有异常会尝试一直消费
#      ack-mode: manual# 手动ack：消费有异常时停止ack-mode: manual_immediate

在这里插入图片描述

1.2、注册异常处理器

package com.atguigu.spring.kafka.consumer.config;
import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.consumer.Consumer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.TopicBuilder;
import org.springframework.kafka.listener.ConsumerAwareListenerErrorHandler;
import org.springframework.kafka.listener.ListenerExecutionFailedException;
import org.springframework.messaging.Message;
@Configuration
public class MyKafkaConfig {@Beanpublic NewTopic springTestPartitionTopic() {return TopicBuilder.name("my_topic1") //主题名称.partitions(3) //分区数量.replicas(3) //副本数量.build();}//方法名就是注入到容器中对象的名称@Beanpublic ConsumerAwareListenerErrorHandler myErrorHandler() {//创建异常处理器：消费者异常时 且注册使用当前异常处理器 会生效return new ConsumerAwareListenerErrorHandler() {@Overridepublic Object handleError(Message<?> message, ListenerExecutionFailedException e, Consumer<?, ?> consumer) {System.out.println("出现异常，消息内容：value = " + message.getPayload());System.out.println("header = "+message.getHeaders());System.out.println("异常信息：" + e.getMessage());System.out.println("=================");return null;}};}
}

1.3、消费者使用异常处理器

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.TopicPartition;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
@Component
public class MyKafkaListenerAck {/*** 自动ack可能会导致漏消息*      spring-kafka：*                     自动ack 如果有异常，会死循环获取消息重新消费*                        不能继续向后消费消息，会导致消息积压**                    手动ack 配置了手动ack，且ack-mode为manual_immediate时，*                        如果消息消费失败，会继续向后消费* @param record*/@KafkaListener(topicPartitions = {@TopicPartition(topic = "my_topic1",partitions = {"0"})}, groupId = "my_group1",errorHandler = "myErrorHandler")public void onMessage1(ConsumerRecord<String, String> record, Acknowledgment acknowledgment) {System.out.println("my_group1消费者获取分区0的消息：topic = "+ record.topic()+"，partition:"+record.partition()+"，offset = "+record.offset()+"，key = "+record.key()+"，value = "+record.value());int i = 1/0;// 手动ack：手动确认消息已经消费 broker 会提交消费者偏移量acknowledgment.acknowledge();}
}

在这里插入图片描述

1.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;@SpringBootTest
class SpringKafkaConsumerApplicationTests {@ResourceKafkaTemplate kafkaTemplate;@Testvoid contextLoads() {for (int i = 0; i < 10; i++) {kafkaTemplate.send("my_topic1",i%3,"", "指定ack-mode: manual_immediate消费"+i);}}}

1.5、创建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);}}

kafka_debug__logbackxml_162">1.6、屏蔽 kafka debug 日志 logback.xml

<configuration>　　　　　　<!-- 如果觉得idea控制台日志太多，src\main\resources目录下新建logback.xml
屏蔽kafka debug --><logger name="org.apache.kafka.clients" level="debug" />
</configuration>

kafka_172">1.7、引入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></dependencies><build><plugins><plugin><groupId>org.springframework.boot</groupId><artifactId>spring-boot-maven-plugin</artifactId></plugin></plugins></build></project>

1.8、消费者控制台：

1.8.1、第一次启动SpringKafkaConsumerApplication

  .   ____          _            __ _ _/\\ / ___'_ __ _ _(_)_ __  __ _ \ \ \ \
( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \\\/  ___)| |_)| | | | | || (_| |  ) ) ) )'  |____| .__|_| |_|_| |_\__, | / / / /=========|_|==============|___/=/_/_/_/:: Spring Boot ::                (v3.0.5)my_group1消费者获取分区0的消息：topic = my_topic1，partition:0，offset = 0，key = ，value = 指定ack-mode: manual_immediate消费0
出现异常，消息内容：value = ConsumerRecord(topic = my_topic1, partition = 0, leaderEpoch = 0, offset = 0, CreateTime = 1717672170845, serialized key size = 0, serialized value size = 39, headers = RecordHeaders(headers = [], isReadOnly = false), key = , value = 指定ack-mode: manual_immediate消费0)
header = {id=91c96206-2381-5fa1-b391-52627056762f, timestamp=1717672488753}
异常信息：Listener method 'public void com.atguigu.spring.kafka.consumer.listener.MyKafkaListenerAck.onMessage1(org.apache.kafka.clients.consumer.ConsumerRecord<java.lang.String, java.lang.String>,org.springframework.kafka.support.Acknowledgment)' threw exception
=================
my_group1消费者获取分区0的消息：topic = my_topic1，partition:0，offset = 1，key = ，value = 指定ack-mode: manual_immediate消费3
出现异常，消息内容：value = ConsumerRecord(topic = my_topic1, partition = 0, leaderEpoch = 0, offset = 1, CreateTime = 1717672170855, serialized key size = 0, serialized value size = 39, headers = RecordHeaders(headers = [], isReadOnly = false), key = , value = 指定ack-mode: manual_immediate消费3)
header = {id=8755d954-615c-37ff-67f0-85521e090b03, timestamp=1717672488753}
异常信息：Listener method 'public void com.atguigu.spring.kafka.consumer.listener.MyKafkaListenerAck.onMessage1(org.apache.kafka.clients.consumer.ConsumerRecord<java.lang.String, java.lang.String>,org.springframework.kafka.support.Acknowledgment)' threw exception
=================
my_group1消费者获取分区0的消息：topic = my_topic1，partition:0，offset = 2，key = ，value = 指定ack-mode: manual_immediate消费6
出现异常，消息内容：value = ConsumerRecord(topic = my_topic1, partition = 0, leaderEpoch = 0, offset = 2, CreateTime = 1717672170855, serialized key size = 0, serialized value size = 39, headers = RecordHeaders(headers = [], isReadOnly = false), key = , value = 指定ack-mode: manual_immediate消费6)
header = {id=68082673-07d2-6f69-3935-c7034a7caf81, timestamp=1717672488753}
异常信息：Listener method 'public void com.atguigu.spring.kafka.consumer.listener.MyKafkaListenerAck.onMessage1(org.apache.kafka.clients.consumer.ConsumerRecord<java.lang.String, java.lang.String>,org.springframework.kafka.support.Acknowledgment)' threw exception
=================
my_group1消费者获取分区0的消息：topic = my_topic1，partition:0，offset = 3，key = ，value = 指定ack-mode: manual_immediate消费9
出现异常，消息内容：value = ConsumerRecord(topic = my_topic1, partition = 0, leaderEpoch = 0, offset = 3, CreateTime = 1717672170855, serialized key size = 0, serialized value size = 39, headers = RecordHeaders(headers = [], isReadOnly = false), key = , value = 指定ack-mode: manual_immediate消费9)
header = {id=bcbda053-9536-df91-d937-2688b5d4c6ea, timestamp=1717672488754}
异常信息：Listener method 'public void com.atguigu.spring.kafka.consumer.listener.MyKafkaListenerAck.onMessage1(org.apache.kafka.clients.consumer.ConsumerRecord<java.lang.String, java.lang.String>,org.springframework.kafka.support.Acknowledgment)' threw exception
=================

1.8.n、第n次启动SpringKafkaConsumerApplication

  .   ____          _            __ _ _/\\ / ___'_ __ _ _(_)_ __  __ _ \ \ \ \
( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \\\/  ___)| |_)| | | | | || (_| |  ) ) ) )'  |____| .__|_| |_|_| |_\__, | / / / /=========|_|==============|___/=/_/_/_/:: Spring Boot ::                (v3.0.5)my_group1消费者获取分区0的消息：topic = my_topic1，partition:0，offset = 0，key = ，value = 指定ack-mode: manual_immediate消费0
出现异常，消息内容：value = ConsumerRecord(topic = my_topic1, partition = 0, leaderEpoch = 0, offset = 0, CreateTime = 1717672170845, serialized key size = 0, serialized value size = 39, headers = RecordHeaders(headers = [], isReadOnly = false), key = , value = 指定ack-mode: manual_immediate消费0)
header = {id=6bc9edb9-ad1c-e00e-9b27-c0c540248091, timestamp=1717672854508}
异常信息：Listener method 'public void com.atguigu.spring.kafka.consumer.listener.MyKafkaListenerAck.onMessage1(org.apache.kafka.clients.consumer.ConsumerRecord<java.lang.String, java.lang.String>,org.springframework.kafka.support.Acknowledgment)' threw exception
=================
my_group1消费者获取分区0的消息：topic = my_topic1，partition:0，offset = 1，key = ，value = 指定ack-mode: manual_immediate消费3
出现异常，消息内容：value = ConsumerRecord(topic = my_topic1, partition = 0, leaderEpoch = 0, offset = 1, CreateTime = 1717672170855, serialized key size = 0, serialized value size = 39, headers = RecordHeaders(headers = [], isReadOnly = false), key = , value = 指定ack-mode: manual_immediate消费3)
header = {id=14e6951b-b25f-10ca-702c-a1699d25645b, timestamp=1717672854508}
异常信息：Listener method 'public void com.atguigu.spring.kafka.consumer.listener.MyKafkaListenerAck.onMessage1(org.apache.kafka.clients.consumer.ConsumerRecord<java.lang.String, java.lang.String>,org.springframework.kafka.support.Acknowledgment)' threw exception
=================
my_group1消费者获取分区0的消息：topic = my_topic1，partition:0，offset = 2，key = ，value = 指定ack-mode: manual_immediate消费6
出现异常，消息内容：value = ConsumerRecord(topic = my_topic1, partition = 0, leaderEpoch = 0, offset = 2, CreateTime = 1717672170855, serialized key size = 0, serialized value size = 39, headers = RecordHeaders(headers = [], isReadOnly = false), key = , value = 指定ack-mode: manual_immediate消费6)
header = {id=26e85ef8-2502-fd29-8d5b-091ec0362900, timestamp=1717672854509}
异常信息：Listener method 'public void com.atguigu.spring.kafka.consumer.listener.MyKafkaListenerAck.onMessage1(org.apache.kafka.clients.consumer.ConsumerRecord<java.lang.String, java.lang.String>,org.springframework.kafka.support.Acknowledgment)' threw exception
=================
my_group1消费者获取分区0的消息：topic = my_topic1，partition:0，offset = 3，key = ，value = 指定ack-mode: manual_immediate消费9
出现异常，消息内容：value = ConsumerRecord(topic = my_topic1, partition = 0, leaderEpoch = 0, offset = 3, CreateTime = 1717672170855, serialized key size = 0, serialized value size = 39, headers = RecordHeaders(headers = [], isReadOnly = false), key = , value = 指定ack-mode: manual_immediate消费9)
header = {id=25c6b6c8-d02d-2091-7c6f-ebaea6c52d1d, timestamp=1717672854509}
异常信息：Listener method 'public void com.atguigu.spring.kafka.consumer.listener.MyKafkaListenerAck.onMessage1(org.apache.kafka.clients.consumer.ConsumerRecord<java.lang.String, java.lang.String>,org.springframework.kafka.support.Acknowledgment)' threw exception
=================

此时如果不关闭SpringKafkaConsumerApplication，生产者继续发送消息，消费者只会往后消费，不会从头再次消费

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


http://www.ppmy.cn/server/48626.html

相关文章

【Python】实现极致：克服PyInstaller打包挑战，解决libpython3.10.so.1.0库丢失难题

【Python】实现极致：克服PyInstaller打包挑战，解决libpython3.10.so.1.0库丢失难题

【Python】实现极致&#xff1a;克服PyInstaller打包挑战&#xff0c;解决libpython3.10.so.1.0库丢失难题 大家好 我是寸铁&#x1f44a; 总结了一篇【Python】实现极致&#xff1a;克服PyInstaller打包挑战&#xff0c;解决libpython3.10.so.1.0库丢失难题✨ 喜欢的小伙伴可以…
阅读更多...
英伟达最新GPU和互联路线图分析

英伟达最新GPU和互联路线图分析

Nvidia在计算、网络和图形领域独树一帜&#xff0c;其显著优势在于雄厚的资金实力及在生成式人工智能市场的领先地位。凭借卓越的架构、工程和供应链&#xff0c;Nvidia能够自由实施创新路线图&#xff0c;引领行业未来。 到 21 世纪&#xff0c;Nvidia 已经是一个非常成功的创…
阅读更多...
Unity 设置默认字体（支持老版及新版TMP）

Unity 设置默认字体（支持老版及新版TMP）

普通UI-Text设置 &#xff08;同一unity版本设置一次即可&#xff09; 1.首先工程的Resources目录下创建Fonts文件夹用于存放字体 如下图所示 2.找到Unity的安装目录下的Editor\Data\Resources\PackageManager\BuiltInPackages\com.unity.ugui\Runtime\UI\Core\Text.cs文件 …
阅读更多...
若依对数据二次处理导致查询total只有十条的问题处理办法

若依对数据二次处理导致查询total只有十条的问题处理办法

前言&#xff1a; 在使用若依框架的过程中&#xff0c;如果是查询结果数据直接返回&#xff0c;那么其自带的分页插件可以正常返回数据以及总条数&#xff0c;若是在业务逻辑层对数据进行了其他二次处理&#xff0c;再返回就会出现异常&#xff0c;无论查询了多少条&#xff0…
阅读更多...
RabbitMQ-Stream(高级详解)

RabbitMQ-Stream(高级详解)

文章目录 什么是流何时使用 RabbitMQ Stream&#xff1f;在 RabbitMQ 中使用流的其他方式基本使用Offset参数chunk Stream 插件服务端消息偏移量追踪示例 示例应用程序RabbitMQ 流 Java API概述环境创建具有所有默认值的环境使用 URI 创建环境创建具有多个 URI 的环境 启用 TLS…
阅读更多...
【数据结构】队列的应用（详解）

【数据结构】队列的应用（详解）

目录 0 引言 1 打印机任务队列 2 广度优先搜索&#xff08;BFS&#xff09; 3 总结 0 引言 队列&#xff08;Queue&#xff09;是一种先进先出&#xff08;FIFO&#xff09;的数据结构&#xff0c;它允许在尾部添加元素&#xff08;入队操作&#xff09;&#xff0c;并…
阅读更多...
DB2问题解决：SQL0270N Function not supported (Reason code = “100“). SQLSTATE=42997

DB2问题解决：SQL0270N Function not supported (Reason code = “100“). SQLSTATE=42997

在执行修改表结构sql语句过程中&#xff0c;报错&#xff1a;SQL0270N Function not supported (Reason code "100"). SQLSTATE42997 。这个错误通常与表的 DATA CAPTURE 设置有关。如果某些表的 DATA CAPTURE 设置为 CHANGES&#xff0c;则某些表结构修改操作可能不…
阅读更多...
Rust基础学习-ModulesPackage

Rust基础学习-ModulesPackage

在Rust中&#xff0c;模块有助于将程序分割成逻辑单元&#xff0c;以提高可读性和组织性。一旦程序变得更大&#xff0c;将其拆分为多个文件或命名空间非常重要。 模块有助于构建我们的程序。模块是项目的集合&#xff1a;包括函数、结构体甚至其他模块。 Module 定义模块 在…
阅读更多...
最新文章

Copyright @ 2022~2023