SpringBoot 处理 @KafkaListener 消息

消息监听容器

1、KafkaMessageListenerContainer

由spring提供用于监听以及拉取消息，并将这些消息按指定格式转换后交给由@KafkaListener注解的方法处理，相当于一个消费者；

看看其整体代码结构：

可以发现其入口方法为doStart(), 往上追溯到实现了SmartLifecycle接口，很明显，由spring管理其start和stop操作；

ListenerConsumer, 内部真正拉取消息消费的是这个结构，其 实现了Runable接口，简言之，它就是一个后台线程轮训拉取并处理消息（while true死循环拉取消息）。

在doStart方法中会创建ListenerConsumer并交给线程池处理

以上步骤就开启了消息监听过程。

KafkaMessageListenerContainer#doStart
protected void doStart() {if (isRunning()) {return;}ContainerProperties containerProperties = getContainerProperties();if (!this.consumerFactory.isAutoCommit()) {AckMode ackMode = containerProperties.getAckMode();if (ackMode.equals(AckMode.COUNT) || ackMode.equals(AckMode.COUNT_TIME)) {Assert.state(containerProperties.getAckCount() > 0, "'ackCount' must be > 0");}if ((ackMode.equals(AckMode.TIME) || ackMode.equals(AckMode.COUNT_TIME))&& containerProperties.getAckTime() == 0) {containerProperties.setAckTime(5000);}}Object messageListener = containerProperties.getMessageListener();Assert.state(messageListener != null, "A MessageListener is required");if (containerProperties.getConsumerTaskExecutor() == null) {SimpleAsyncTaskExecutor consumerExecutor = new SimpleAsyncTaskExecutor((getBeanName() == null ? "" : getBeanName()) + "-C-");containerProperties.setConsumerTaskExecutor(consumerExecutor);}Assert.state(messageListener instanceof GenericMessageListener, "Listener must be a GenericListener");this.listener = (GenericMessageListener<?>) messageListener;ListenerType listenerType = ListenerUtils.determineListenerType(this.listener);if (this.listener instanceof DelegatingMessageListener) {Object delegating = this.listener;while (delegating instanceof DelegatingMessageListener) {delegating = ((DelegatingMessageListener<?>) delegating).getDelegate();}listenerType = ListenerUtils.determineListenerType(delegating);}// 这里创建了监听消费者对象this.listenerConsumer = new ListenerConsumer(this.listener, listenerType);setRunning(true);// 将消费者对象放入到线程池中执行this.listenerConsumerFuture = containerProperties.getConsumerTaskExecutor().submitListenable(this.listenerConsumer);}
KafkaMessageListenerContainer.ListenerConsumer#run
public void run() {this.consumerThread = Thread.currentThread();if (this.genericListener instanceof ConsumerSeekAware) {((ConsumerSeekAware) this.genericListener).registerSeekCallback(this);}if (this.transactionManager != null) {ProducerFactoryUtils.setConsumerGroupId(this.consumerGroupId);}this.count = 0;this.last = System.currentTimeMillis();if (isRunning() && this.definedPartitions != null) {try {initPartitionsIfNeeded();}catch (Exception e) {this.logger.error("Failed to set initial offsets", e);}}long lastReceive = System.currentTimeMillis();long lastAlertAt = lastReceive;while (isRunning()) {try {if (!this.autoCommit && !this.isRecordAck) {processCommits();}processSeeks();if (!this.consumerPaused && isPaused()) {this.consumer.pause(this.consumer.assignment());this.consumerPaused = true;if (this.logger.isDebugEnabled()) {this.logger.debug("Paused consumption from: " + this.consumer.paused());}publishConsumerPausedEvent(this.consumer.assignment());}// 拉取信息ConsumerRecords<K, V> records = this.consumer.poll(this.containerProperties.getPollTimeout());this.lastPoll = System.currentTimeMillis();if (this.consumerPaused && !isPaused()) {if (this.logger.isDebugEnabled()) {this.logger.debug("Resuming consumption from: " + this.consumer.paused());}Set<TopicPartition> paused = this.consumer.paused();this.consumer.resume(paused);this.consumerPaused = false;publishConsumerResumedEvent(paused);}if (records != null && this.logger.isDebugEnabled()) {this.logger.debug("Received: " + records.count() + " records");if (records.count() > 0 && this.logger.isTraceEnabled()) {this.logger.trace(records.partitions().stream().flatMap(p -> records.records(p).stream())// map to same format as send metadata toString().map(r -> r.topic() + "-" + r.partition() + "@" + r.offset()).collect(Collectors.toList()));}}if (records != null && records.count() > 0) {if (this.containerProperties.getIdleEventInterval() != null) {lastReceive = System.currentTimeMillis();}invokeListener(records);}else {if (this.containerProperties.getIdleEventInterval() != null) {long now = System.currentTimeMillis();if (now > lastReceive + this.containerProperties.getIdleEventInterval()&& now > lastAlertAt + this.containerProperties.getIdleEventInterval()) {publishIdleContainerEvent(now - lastReceive, this.isConsumerAwareListener? this.consumer : null, this.consumerPaused);lastAlertAt = now;if (this.genericListener instanceof ConsumerSeekAware) {seekPartitions(getAssignedPartitions(), true);}}}}}catch (WakeupException e) {// Ignore, we're stopping}catch (NoOffsetForPartitionException nofpe) {this.fatalError = true;ListenerConsumer.this.logger.error("No offset and no reset policy", nofpe);break;}catch (Exception e) {handleConsumerException(e);}}ProducerFactoryUtils.clearConsumerGroupId();if (!this.fatalError) {if (this.kafkaTxManager == null) {commitPendingAcks();try {this.consumer.unsubscribe();}catch (WakeupException e) {// No-op. Continue process}}}else {ListenerConsumer.this.logger.error("No offset and no reset policy; stopping container");KafkaMessageListenerContainer.this.stop();}this.monitorTask.cancel(true);if (!this.taskSchedulerExplicitlySet) {((ThreadPoolTaskScheduler) this.taskScheduler).destroy();}this.consumer.close();this.logger.info("Consumer stopped");}

2、ConcurrentMessageListenerContainer

并发消息监听，相当于创建消费者；其底层逻辑仍然是通过KafkaMessageListenerContainer实现处理；从实现上看就是在KafkaMessageListenerContainer上做了层包装，有多少的concurrency就创建多个KafkaMessageListenerContainer，也就是concurrency个消费者。

	protected void doStart() {if (!isRunning()) {ContainerProperties containerProperties = getContainerProperties();TopicPartitionInitialOffset[] topicPartitions = containerProperties.getTopicPartitions();if (topicPartitions != null&& this.concurrency > topicPartitions.length) {this.logger.warn("When specific partitions are provided, the concurrency must be less than or "+ "equal to the number of partitions; reduced from " + this.concurrency + " to "+ topicPartitions.length);this.concurrency = topicPartitions.length;}setRunning(true);// 创建多个消费者for (int i = 0; i < this.concurrency; i++) {KafkaMessageListenerContainer<K, V> container;if (topicPartitions == null) {container = new KafkaMessageListenerContainer<>(this, this.consumerFactory,containerProperties);}else {container = new KafkaMessageListenerContainer<>(this, this.consumerFactory,containerProperties, partitionSubset(containerProperties, i));}String beanName = getBeanName();container.setBeanName((beanName != null ? beanName : "consumer") + "-" + i);if (getApplicationEventPublisher() != null) {container.setApplicationEventPublisher(getApplicationEventPublisher());}container.setClientIdSuffix("-" + i);container.setAfterRollbackProcessor(getAfterRollbackProcessor());container.start();this.containers.add(container);}}}

3、@KafkaListener底层监听原理

上面已经介绍了KafkaMessageListenerContainer的作用是拉取并处理消息，但还缺少关键的一步，即 如何将我们的业务逻辑与KafkaMessageListenerContainer的处理逻辑联系起来？

那么这个桥梁就是@KafkaListener注解

KafkaListenerAnnotationBeanPostProcessor, 从后缀BeanPostProcessor就可以知道这是Spring IOC初始化bean相关的操作，当然这里也是；此类会扫描带@KafkaListener注解的类或者方法，通过 KafkaListenerContainerFactory工厂创建对应的KafkaMessageListenerContainer，并调用start方法启动监听，也就是这样打通了这条路…

4、Spring Boot 自动加载kafka相关配置

1、KafkaAutoConfiguration
自动生成kafka相关配置，比如当缺少这些bean的时候KafkaTemplate、ProducerListener、ConsumerFactory、ProducerFactory等，默认创建bean实例

2、KafkaAnnotationDrivenConfiguration
主要是针对于spring-kafka提供的注解背后的相关操作，比如 @KafkaListener;

在开启了@EnableKafka注解后，spring会扫描到此配置并创建缺少的bean实例，比如当配置的工厂beanName不是kafkaListenerContainerFactory的时候，就会默认创建一个beanName为kafkaListenerContainerFactory的实例，这也是为什么在springboot中不用定义consumer的相关配置也可以通过@KafkaListener正常的处理消息

5、消息处理

1、单条消息处理

@Configuration
public class KafkaConsumerConfiguration {@BeanKafkaListenerContainerFactory<ConcurrentMessageListenerContainer<Integer, String>> kafkaCustomizeContainerFactory() {ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();factory.setConsumerFactory(consumerFactory());factory.setConcurrency(2);factory.getContainerProperties().setPollTimeout(3000);return factory;}private ConsumerFactory<Integer, String> consumerFactory() {return new DefaultKafkaConsumerFactory<>(consumerConfigs());}private Map<String, Object> consumerConfigs() {Map<String, Object> props = new HashMap<>();props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bizConfig.getReconciliationInstanceKafkaServers());props.put(ConsumerConfig.GROUP_ID_CONFIG, bizConfig.getReconciliationInstanceKafkaConsumerGroupId());props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "latest");props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true);props.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, 100);props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);props.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, 100);props.put(ConsumerConfig.FETCH_MIN_BYTES_CONFIG, 300);// poll 一次拉取的阻塞的最大时长，单位：毫秒。这里指的是阻塞拉取需要满足至少 fetch-min-size 大小的消息props.put(ConsumerConfig.FETCH_MAX_BYTES_CONFIG, 10000);return props;}}

这种方式的@KafkaLisener中的参数是单条的。

2、批量处理

@Configuration
@EnableKafka
public class KafkaConfig {@Bean
public KafkaListenerContainerFactory<?, ?> batchFactory() {ConcurrentKafkaListenerContainerFactory<Integer, String> factory =new ConcurrentKafkaListenerContainerFactory<>();factory.setConsumerFactory(consumerFactory());// 增加开启批量处理factory.setBatchListener(true);  // <<<<<<<<<<<<<<<<<<<<<<<<<return factory;
}@Beanpublic ConsumerFactory<Integer, String> consumerFactory() {return new DefaultKafkaConsumerFactory<>(consumerConfigs());}@Beanpublic Map<String, Object> consumerConfigs() {Map<String, Object> props = new HashMap<>();props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, embeddedKafka.getBrokersAsString());...return props;}
}// 注意：这里接受的是集合类型
@KafkaListener(id = "list", topics = "myTopic", containerFactory = "batchFactory")
public void listen(List<String> list) {...
}

这种方式的@KafkaLisener中的参数是多条的。

6、线程池相关

如果没有额外给Kafka指定线程池，底层默认用的是SimpleAsyncTaskExecutor类，它不使用线程池，而是为每个任务创建新线程。相当于一个消费者用一个独立的线程来跑。

总结

spring为了将kafka融入其生态，方便在spring大环境下使用kafka，开发了spring-kafa这一模块，本质上是为了帮助开发者更好的以spring的方式使用kafka

@KafkaListener就是这么一个工具，在同一个项目中既可以有单条的消息处理，也可以配置多条的消息处理，稍微改变下配置即可实现，很是方便

当然，@KafkaListener单条或者多条消息处理仍然是spring自行封装处理，与kafka-client客户端的拉取机制无关；比如一次性拉取50条消息，对于单条处理来说就是循环50次处理，而多条消息处理则可以一次性处理50条；本质上来说这套逻辑都是spring处理的，并不是说单条消费就是通过kafka-client一次只拉取一条消息

在使用过程中需要注意spring自动的创建的一些bean实例，当然也可以覆盖其自动创建的实例以满足特定的需求场景。

                        
原文链接：https://blog.csdn.net/yuechuzhixing/article/details/124725713

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

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

相关文章

Linux 访问控制列表（Access Control List）

Linux 访问控制列表（Access Control List）

在Linux中&#xff0c;目录或文件的权限是针对的所有者(owner)&#xff0c;所属组(group)&#xff0c;其他人(others)这3种类别来设置的。这种根据类别控制权限的方法无法精确控制每个用户的行为。为了解决这个问题&#xff0c;Linux引入了访问控制列表&#xff08;Access Cont…
阅读更多...
Navicat 17 新特性 | 聚焦 MongoDB

Navicat 17 新特性 | 聚焦 MongoDB

随着 Navicat 17 的盛大发布&#xff0c;其一系列创新特性赢得了广大用户的热烈反响。它不仅在模型设计上实现了突破性优化&#xff0c;提升了查询与配置的效率&#xff0c;还大幅优化了用户界面的交互体验&#xff0c;原生支持国产平台与操作系统&#xff0c;同时增强 BI 能力…
阅读更多...
JAVA毕业设计170—基于Java+Springboot+vue3+小程序的房屋租赁小程序系统(源代码+数据库)

JAVA毕业设计170—基于Java+Springboot+vue3+小程序的房屋租赁小程序系统(源代码+数据库)

毕设所有选题&#xff1a; https://blog.csdn.net/2303_76227485/article/details/131104075 基于JavaSpringbootvue3小程序的房屋租赁小程序系统(源代码数据库)170 一、系统介绍 本项目前后端分离(可以改为ssm版本)&#xff0c;分为用户、房东、管理员三种角色 1、用户&am…
阅读更多...
Qt+FFmpeg开发视频播放器笔记(三)：音视频流解析封装

Qt+FFmpeg开发视频播放器笔记(三)：音视频流解析封装

音频解析 音频解码是指将压缩的音频数据转换为可以再生的PCM（脉冲编码调制）数据的过程。 FFmpeg音频解码的基本步骤如下： 初始化FFmpeg解码器(4.0版本后可省略): 调用av_register_all()初始化编解码器。 调用avcodec_register_all()注册所有编解码器。 打开输入的音频流:…
阅读更多...
Idea springboot项目热部署

Idea springboot项目热部署

使用 spring-boot-devtools spring-boot-devtools 是 Spring Boot 提供的开发工具模块&#xff0c;它可以自动检测到代码的变化并重启应用&#xff0c;实现热部署。 配置步骤&#xff1a; 添加依赖&#xff1a; 在项目的 pom.xml 中加入 spring-boot-devtools 依赖&#xff1…
阅读更多...
Redis（主从复制、哨兵模式、集群）概述及部署测试

Redis（主从复制、哨兵模式、集群）概述及部署测试

目录 一、Redis 主从复制 1.1、Redis 主从复制概念 1.2、主从复制的作用 1.3、主从复制流程 1.4、搭建Redis 主从复制 二、Redis 哨兵模式 2.1、Redis 哨兵模式概念 2.2、哨兵模式原理 2.3、哨兵模式的作用 2.4、哨兵模式的结构 2.5、故障转移机制 2.6、主节点的选…
阅读更多...
Node.js 多版本安装与切换指南

Node.js 多版本安装与切换指南

一.使用nvm的方法 1. 卸载nodejs 如果你的电脑有安装nodejs&#xff0c;需要先卸载掉&#xff1b;若没有请直接下一步。 2. 前往官网下载nvm nvm&#xff1a;一个nodejs版本管理工具&#xff01; 官网地址&#xff1a;nvm文档手册 - nvm是一个nodejs版本管理工具 - nvm中文…
阅读更多...
MySQL详解：数据类型、约束

MySQL详解：数据类型、约束

MySQL 1. 数据类型1.1 数值类型1.1.1 bit 位类型1.1.2 整数数据类型1.1.3 小数类型floatdecimal 1.2 字符类型1.2.1 char1.2.2 varchar 可变长字符串1.2.3 日期和时间类型datedatetimetimestamp 1.2.4 enum1.2.5 set集合查询函数 find_in_set 2. 表的约束2.1 NULL 空属性2.2 默…
阅读更多...
基于鸿蒙API10的RTSP播放器(七：亮度调节功能测试）

基于鸿蒙API10的RTSP播放器(七：亮度调节功能测试）

目标&#xff1a; 当我的手指在设备左方进行上下移动的时候&#xff0c;可以进行屏幕亮度的调节&#xff0c;在调节的同时&#xff0c;有实时的调节进度条显示 步骤&#xff1a; 界面逻辑&#xff1a;使用Stack() 组件&#xff0c;完成音量图标和进度条的组合显示&#xff0c…
阅读更多...
Linux echo，printf 命令

Linux echo，printf 命令

参考资料 【Linux】ハイフンをいっぱい出したかっただけなのに【printfコマンド】 目录 一. echo命令1.1 -n 选项1.2 -e 选项1.3 配合扩展实现批量换行输出1.3.1 xargs -n 11.3.2 tr \n1.3.3 xargs printf "%s\n"1.4 ANSI转义序列1.5 彩色文本输出 二. printf 命令…
阅读更多...
C# System.BadImageFormatException问题及解决

C# System.BadImageFormatException问题及解决

C# System.BadImageFormatException问题 出现System.BadImageFormatException 异常有两种情况&#xff1a;程序目标平台不一致&引用dll文件的系统平台不一致。 异常参考 BadImageFormatException 程序目标平台不一致&#xff1a; 项目>属性>生成&#xff1a;x86 …
阅读更多...
【吊打面试官系列-Redis面试题】使用过 Redis 做异步队列么，你是怎么用的？

【吊打面试官系列-Redis面试题】使用过 Redis 做异步队列么，你是怎么用的？

大家好&#xff0c;我是锋哥。今天分享关于【使用过 Redis 做异步队列么&#xff0c;你是怎么用的&#xff1f;】面试题&#xff0c;希望对大家有帮助&#xff1b; 使用过 Redis 做异步队列么&#xff0c;你是怎么用的&#xff1f; 一般使用 list 结构作为队列&#xff0c;rpus…
阅读更多...
【Redis】redis5种数据类型（list）

【Redis】redis5种数据类型（list）

目录 基本介绍 命令 LPUSH LPUSHX RPUSH RPUSHX LRANGE LPOP RPOP LINDEX LINSERT LLEN LREM LTRIM LSET 阻塞版本的命令 BLPOP 内部编码 基本介绍 list相当于c的双端队列deque 区分获取和删除的区别 lindex能获取到元素的值lrem也能返回被删除元素的值 命…
阅读更多...
828华为云征文 | 华为云FlexusX实例下的Kafka集群部署实践与性能优化

828华为云征文 | 华为云FlexusX实例下的Kafka集群部署实践与性能优化

前言 华为云FlexusX实例&#xff0c;以创新的柔性算力技术&#xff0c;为Kafka集群部署带来前所未有的性能飞跃。其灵活的CPU与内存配比&#xff0c;结合智能调度与加速技术&#xff0c;让Kafka在高并发场景下依然游刃有余。在828华为云企业上云节期间&#xff0c;FlexusX实例携…
阅读更多...
手机玩机常识____展讯芯片刷机平台ResearchDownload的一些基本常识与问题解决

手机玩机常识____展讯芯片刷机平台ResearchDownload的一些基本常识与问题解决

展讯ResearchDownload工具 展讯芯片的刷机工具--ResearchDownload下载工具"是一款专为用户设计的高效、便捷的下载管理软件&#xff0c;它能够帮助用户快速、稳定地从互联网上获取各种文件。这款工具以其强大的功能和良好的用户体验&#xff0c;在众多展讯芯片下载工具中脱…
阅读更多...
git-describe获取不到新创建的标签

git-describe获取不到新创建的标签

一、问题描述 1、新建的分支 2、git-describe 失败 二、查询资料 &#xff08;1&#xff09;git-describe - 根据可用的ref给对象一个人类可读的名称 &#xff08;2&#xff09;该命令查找可从提交访问的最新标记。如果标记指向提交&#xff0c;则仅显示标记。否则&#xf…
阅读更多...
S-Procedure的基本形式及使用

S-Procedure的基本形式及使用

理论 Lemma 1. ( S- Procedure[ 34] ) : Define the quadratic func- \textbf{Lemma 1. ( S- Procedure[ 34] ) : Define the quadratic func- } Lemma 1. ( S- Procedure[ 34] ) : Define the quadratic func- tions w.r.t. x ∈ C M 1 \mathbf{x}\in\mathbb{C}^M\times1 x…
阅读更多...
el-input设置type=‘number‘和v-model.number的区别

el-input设置type=‘number‘和v-model.number的区别

el-input设置typenumber’与设置.number修饰符的区别 1. 设置type‘number’ 使用el-input时想收集数字类型的数据&#xff0c;我们首先会想到typenumber&#xff0c;设置完type为number时会限制我们输入的内容只能为数字&#xff0c;不能为字符/汉字等非数字类型的数值&…
阅读更多...
Leetcode面试经典150题-148.排序链表

Leetcode面试经典150题-148.排序链表

题目比较简单&#xff0c;使用链表的归并排序 解法都在代码里&#xff0c;不懂就留言或者私信 合并链表部分没怎么加注释&#xff0c;时间实在是不充裕&#xff0c;看不懂的看一下这篇专门讲解合并链表的 Leetcode面试经典150题-21.合并两个有序链表-CSDN博客 /*** Definit…
阅读更多...
Brave编译指南2024 Windows篇：安装Visual Studio 2022（二）

Brave编译指南2024 Windows篇：安装Visual Studio 2022（二）

1.引言 在编译Brave浏览器之前&#xff0c;安装和配置合适的开发工具是至关重要的一步。Visual Studio 2022是编译Brave浏览器所需的重要开发环境&#xff0c;它提供了一整套工具和服务&#xff0c;以支持多种编程语言和技术。作为一款功能强大的集成开发环境&#xff08;IDE&…
阅读更多...
最新文章
推荐文章

Copyright @ 2022~2023

友情链接： 我的编程经验分享 一条长河网 尧图网站开发