springboot整合kafka-笔记

springboot整合kafka-笔记

配置pom.xml

这里我的springboot版本是2.3.8.RELEASE，使用的kafka-mq的版本是2.12

<dependencyManagement><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-dependencies</artifactId><version>2.3.8.RELEASE</version><type>pom</type><scope>import</scope></dependency><dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId><version>2.3.6.RELEASE</version></dependency><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>2.3.1</version></dependency></dependencies></dependencyManagement><dependencies><dependency><groupId>org.springframework.boot</groupId><artifactId>spring-boot-starter-web</artifactId></dependency><dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId><version>2.3.6.RELEASE</version></dependency><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>2.3.1</version></dependency></dependencies>

配置application.yml

spring:kafka:bootstrap-servers: 10.1.5.212:9092

配置EnableKafka注解

@EnableKafka
@EnableScheduling
@SpringBootApplication
public class DemoApplication {public static void main(String[] args) {SpringApplication.run(DemoApplication.class,args);}
}

定义kafka消息生产者

package cn.test.kafka;import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
@Slf4j
@Service
public class KafkaProducerService {@Autowiredprivate  KafkaTemplate<String, String> kafkaTemplate;private  String topic= "my816topic";public void sendMessage(String message){log.info("向kafka发送消息:{}",message);kafkaTemplate.send(topic,message);}
}

定义kafka消息消费者

package cn.test.kafka;import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Service;@Slf4j
@Service
public class KafkaConsumerService {@KafkaListener(topics = "my816topic",groupId = "mygroup")public void listen(String message){log.info("rec msg:{}",message);}
}

测试发送kafka消息

@Component
@Slf4j
public class TestJob {@Autowiredprivate ObjectMapper objectMapper;@Autowiredprivate KafkaProducerService kafkaProducerService;@Scheduled(cron = "0/13 * * * * ?")public void hwreg() {String s = null;try {s = objectMapper.writeValueAsString(new CityInfo("hefei"+System.currentTimeMillis(), 117.17, 31.52));} catch (JsonProcessingException e) {e.printStackTrace();}//测试发送字符串消息kafkaProducerService.sendMessage(s);}

测试发送kafka消息-控制台日志

2023-08-16 19:16:50,832  INFO [17324]  [main] [] o.a.k.c.c.ConsumerConfig [AbstractConfig.java : 347] ConsumerConfig values: allow.auto.create.topics = trueauto.commit.interval.ms = 5000auto.offset.reset = latestbootstrap.servers = [10.1.5.212:9092]check.crcs = trueclient.dns.lookup = defaultclient.id = client.rack = connections.max.idle.ms = 540000default.api.timeout.ms = 60000enable.auto.commit = falseexclude.internal.topics = truefetch.max.bytes = 52428800fetch.max.wait.ms = 500fetch.min.bytes = 1group.id = mygroupgroup.instance.id = nullheartbeat.interval.ms = 3000interceptor.classes = []internal.leave.group.on.close = trueisolation.level = read_uncommittedkey.deserializer = class org.apache.kafka.common.serialization.StringDeserializermax.partition.fetch.bytes = 1048576max.poll.interval.ms = 300000max.poll.records = 500metadata.max.age.ms = 300000metric.reporters = []metrics.num.samples = 2metrics.recording.level = INFOmetrics.sample.window.ms = 30000partition.assignment.strategy = [class org.apache.kafka.clients.consumer.RangeAssignor]receive.buffer.bytes = 65536reconnect.backoff.max.ms = 1000reconnect.backoff.ms = 50request.timeout.ms = 30000retry.backoff.ms = 100sasl.client.callback.handler.class = nullsasl.jaas.config = nullsasl.kerberos.kinit.cmd = /usr/bin/kinitsasl.kerberos.min.time.before.relogin = 60000sasl.kerberos.service.name = nullsasl.kerberos.ticket.renew.jitter = 0.05sasl.kerberos.ticket.renew.window.factor = 0.8sasl.login.callback.handler.class = nullsasl.login.class = nullsasl.login.refresh.buffer.seconds = 300sasl.login.refresh.min.period.seconds = 60sasl.login.refresh.window.factor = 0.8sasl.login.refresh.window.jitter = 0.05sasl.mechanism = GSSAPIsecurity.protocol = PLAINTEXTsend.buffer.bytes = 131072session.timeout.ms = 10000ssl.cipher.suites = nullssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]ssl.endpoint.identification.algorithm = httpsssl.key.password = nullssl.keymanager.algorithm = SunX509ssl.keystore.location = nullssl.keystore.password = nullssl.keystore.type = JKSssl.protocol = TLSssl.provider = nullssl.secure.random.implementation = nullssl.trustmanager.algorithm = PKIXssl.truststore.location = nullssl.truststore.password = nullssl.truststore.type = JKSvalue.deserializer = class org.apache.kafka.common.serialization.StringDeserializer2023-08-16 19:16:50,884  INFO [17324]  [main] [] o.a.k.c.u.AppInfoParser [AppInfoParser.java : 117] Kafka version: 2.3.1
2023-08-16 19:16:50,884  INFO [17324]  [main] [] o.a.k.c.u.AppInfoParser [AppInfoParser.java : 118] Kafka commitId: 18a913733fb71c01
2023-08-16 19:16:50,884  INFO [17324]  [main] [] o.a.k.c.u.AppInfoParser [AppInfoParser.java : 119] Kafka startTimeMs: 1692184610882
2023-08-16 19:16:50,887  INFO [17324]  [main] [] o.a.k.c.c.KafkaConsumer [KafkaConsumer.java : 964] [Consumer clientId=consumer-1, groupId=mygroup] Subscribed to topic(s): my816topic
2023-08-16 19:16:51,094  INFO [17324]  [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] [] o.a.k.c.Metadata [Metadata.java : 261] [Consumer clientId=consumer-1, groupId=mygroup] Cluster ID: mUe0Z0StRd2_X-_m_S557A
2023-08-16 19:16:51,096  INFO [17324]  [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] [] o.a.k.c.c.i.AbstractCoordinator [AbstractCoordinator.java : 728] [Consumer clientId=consumer-1, groupId=mygroup] Discovered group coordinator 10.1.5.212:9092 (id: 2147483647 rack: null)
2023-08-16 19:16:51,099  INFO [17324]  [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] [] o.a.k.c.c.i.ConsumerCoordinator [ConsumerCoordinator.java : 476] [Consumer clientId=consumer-1, groupId=mygroup] Revoking previously assigned partitions []
2023-08-16 19:16:51,099  INFO [17324]  [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] [] o.s.k.l.KafkaMessageListenerContainer [LogAccessor.java : 292] mygroup: partitions revoked: []
2023-08-16 19:16:51,100  INFO [17324]  [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] [] o.a.k.c.c.i.AbstractCoordinator [AbstractCoordinator.java : 505] [Consumer clientId=consumer-1, groupId=mygroup] (Re-)joining group
2023-08-16 19:16:51,114  INFO [17324]  [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] [] o.a.k.c.c.i.AbstractCoordinator [AbstractCoordinator.java : 505] [Consumer clientId=consumer-1, groupId=mygroup] (Re-)joining group
2023-08-16 19:16:51,126  INFO [17324]  [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] [] o.a.k.c.c.i.AbstractCoordinator [AbstractCoordinator.java : 469] [Consumer clientId=consumer-1, groupId=mygroup] Successfully joined group with generation 9
2023-08-16 19:16:51,130  INFO [17324]  [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] [] o.a.k.c.c.i.ConsumerCoordinator [ConsumerCoordinator.java : 283] [Consumer clientId=consumer-1, groupId=mygroup] Setting newly assigned partitions: my816topic-0
2023-08-16 19:16:51,141  INFO [17324]  [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] [] o.a.k.c.c.i.ConsumerCoordinator [ConsumerCoordinator.java : 525] [Consumer clientId=consumer-1, groupId=mygroup] Setting offset for partition my816topic-0 to the committed offset FetchPosition{offset=288, offsetEpoch=Optional.empty, currentLeader=LeaderAndEpoch{leader=10.1.5.212:9092 (id: 0 rack: null), epoch=0}}
2023-08-16 19:16:51,152  INFO [17324]  [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] [] o.s.k.l.KafkaMessageListenerContainer [LogAccessor.java : 292] mygroup: partitions assigned: [my816topic-0]
2023-08-16 19:16:52,049  INFO [17324]  [scheduling-1] [] c.t.k.KafkaProducerService [KafkaProducerService.java : 19] 向kafka发送消息:{"city":"hefei1692184612007","longitude":117.17,"latitude":31.52}
2023-08-16 19:16:52,053  INFO [17324]  [scheduling-1] [] o.a.k.c.p.ProducerConfig [AbstractConfig.java : 347] ProducerConfig values: acks = 1batch.size = 16384bootstrap.servers = [10.1.5.212:9092]buffer.memory = 33554432client.dns.lookup = defaultclient.id = compression.type = noneconnections.max.idle.ms = 540000delivery.timeout.ms = 120000enable.idempotence = falseinterceptor.classes = []key.serializer = class org.apache.kafka.common.serialization.StringSerializerlinger.ms = 0max.block.ms = 60000max.in.flight.requests.per.connection = 5max.request.size = 1048576metadata.max.age.ms = 300000metric.reporters = []metrics.num.samples = 2metrics.recording.level = INFOmetrics.sample.window.ms = 30000partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitionerreceive.buffer.bytes = 32768reconnect.backoff.max.ms = 1000reconnect.backoff.ms = 50request.timeout.ms = 30000retries = 2147483647retry.backoff.ms = 100sasl.client.callback.handler.class = nullsasl.jaas.config = nullsasl.kerberos.kinit.cmd = /usr/bin/kinitsasl.kerberos.min.time.before.relogin = 60000sasl.kerberos.service.name = nullsasl.kerberos.ticket.renew.jitter = 0.05sasl.kerberos.ticket.renew.window.factor = 0.8sasl.login.callback.handler.class = nullsasl.login.class = nullsasl.login.refresh.buffer.seconds = 300sasl.login.refresh.min.period.seconds = 60sasl.login.refresh.window.factor = 0.8sasl.login.refresh.window.jitter = 0.05sasl.mechanism = GSSAPIsecurity.protocol = PLAINTEXTsend.buffer.bytes = 131072ssl.cipher.suites = nullssl.enabled.protocols = [TLSv1.2, TLSv1.1, TLSv1]ssl.endpoint.identification.algorithm = httpsssl.key.password = nullssl.keymanager.algorithm = SunX509ssl.keystore.location = nullssl.keystore.password = nullssl.keystore.type = JKSssl.protocol = TLSssl.provider = nullssl.secure.random.implementation = nullssl.trustmanager.algorithm = PKIXssl.truststore.location = nullssl.truststore.password = nullssl.truststore.type = JKStransaction.timeout.ms = 60000transactional.id = nullvalue.serializer = class org.apache.kafka.common.serialization.StringSerializer2023-08-16 19:16:52,072  INFO [17324]  [scheduling-1] [] o.a.k.c.u.AppInfoParser [AppInfoParser.java : 117] Kafka version: 2.3.1
2023-08-16 19:16:52,072  INFO [17324]  [scheduling-1] [] o.a.k.c.u.AppInfoParser [AppInfoParser.java : 118] Kafka commitId: 18a913733fb71c01
2023-08-16 19:16:52,072  INFO [17324]  [scheduling-1] [] o.a.k.c.u.AppInfoParser [AppInfoParser.java : 119] Kafka startTimeMs: 1692184612071
2023-08-16 19:16:52,079  INFO [17324]  [kafka-producer-network-thread | producer-1] [] o.a.k.c.Metadata [Metadata.java : 261] [Producer clientId=producer-1] Cluster ID: mUe0Z0StRd2_X-_m_S557A2023-08-16 19:16:52,115  INFO [17324]  [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] [] c.t.k.KafkaConsumerService [KafkaConsumerService.java : 14] rec msg:{"city":"hefei1692184612007","longitude":117.17,"latitude":31.52}

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

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

相关文章

第7步---MySQL的视图操作和

第7步---MySQL的视图操作和

第7步---MySQL的视图操作 虚拟表。保存的只是视图的定义。不存放真实的数据&#xff0c;数据还是在原先的表中。 好处是方便和简化代码以及安全。 1.视图创建 数据准备 -- 创建表的测试数据 create table dept(deptno int primary key,dname varchar(20),loc varchar(20) ); …
阅读更多...
Android学习之路(8) Activity

Android学习之路(8) Activity

本节引言&#xff1a; 本节开始讲解Android的四大组件之一的Activity(活动)&#xff0c;先来看下官方对于Activity的介绍&#xff1a; 移动应用体验与桌面体验的不同之处在于&#xff0c;用户与应用的互动并不总是在同一位置开始&#xff0c;而是经常以不确定的方式开始。例如&…
阅读更多...
项目管理实战笔记1：项目管理常识

项目管理实战笔记1：项目管理常识

序 看了下极客时间的《项目管理实战》&#xff0c;觉得跟之前学习PMP的标准资料还是有所侧重。重新整理下&#xff0c;相比书上繁杂的知识&#xff0c;这个更通俗易懂。 1 角色转换&#xff1a;三大误区 误区1&#xff1a;事必躬亲 自己做事情是可控的&#xff0c;做项目依赖…
阅读更多...
【脚踢数据结构】图（纯享版）

【脚踢数据结构】图（纯享版）

(꒪ꇴ꒪ )&#xff0c;Hello我是祐言QAQ我的博客主页&#xff1a;C/C语言&#xff0c;Linux基础&#xff0c;ARM开发板&#xff0c;软件配置等领域博主&#x1f30d;快上&#x1f698;&#xff0c;一起学习&#xff0c;让我们成为一个强大的攻城狮&#xff01;送给自己和读者的…
阅读更多...
使用 takeUntil 操作符管理 Angular 组件的订阅

使用 takeUntil 操作符管理 Angular 组件的订阅

在 Rxjs 中&#xff0c;可以使用 takeUntil 来控制另外一个 Observable 对象数据的产生。使用 takeUntil&#xff0c;上游的数据直接转手给下游&#xff0c;直到takeUntil的参数吐出一个数据或者完结。 就像一个水龙头开关&#xff0c;一开始是打开的状态&#xff0c;上游的数…
阅读更多...
Jtti：Ubuntu怎么限制指定端口和IP访问

Jtti：Ubuntu怎么限制指定端口和IP访问

在 Ubuntu 系统中&#xff0c;可以使用防火墙规则来限制特定的端口和IP访问。常用的防火墙管理工具是 iptables&#xff0c;以下是使用 iptables 来限制指定端口和IP访问的步骤&#xff1a; 安装 iptables&#xff1a; 如果系统中没有安装 iptables&#xff0c;可以使用以下命…
阅读更多...
Python Django 模型概述与应用

Python Django 模型概述与应用

今天来为大家介绍 Django 框架的模型部分&#xff0c;模型是真实数据的简单明确的描述&#xff0c;它包含了储存的数据所必要的字段和行为&#xff0c;Django 遵循 DRY Principle 。它的目标是你只需要定义数据模型&#xff0c;然后其它的杂七杂八代码你都不用关心&#xff0c;…
阅读更多...
“Spring管理JavaBean的过程及Bean的生命周期“

“Spring管理JavaBean的过程及Bean的生命周期“

目录 引言1.弹簧容器2. Bean的生命周期2.1 配置javaBean2.2. 解析Bean的定义2.3 检查是否需要添加自己的功能2.4 初始化2.5 实现Aware接口2.6 扩展2.7. 销毁 3. 单例模式和原型模式3.1. 单例模式3.2. 原型模式 4. 总结 引言 Spring框架是一个非常流行的Java应用程序框架&#…
阅读更多...
【前端】快速掌握HTML+CSS核心知识点

【前端】快速掌握HTML+CSS核心知识点

文章目录 1.HTML核心基础知识1.1.编写第一个HTML网页1.2.超链接a标签和路径1.3.图像img标签的用法1.4.表格table标签用法1.5.列表ul、ol、dl标签用法1.6.表单form标签用法1.7.区块标签和行内标签用法 2.CSS核心基础知识2.1.CSS标签选择器viewport布局2.2.CSS样式的几种写法2.3.…
阅读更多...
vue 组件-单文件组件

vue 组件-单文件组件

Vue的单文件组件&#xff08;Single File Component&#xff09;是一种将模板、样式和逻辑代码封装在一个文件中的组件开发方式。它使用了以.vue为后缀的文件&#xff0c;结构清晰&#xff0c;便于维护和复用。 示例 一个典型的Vue单文件组件包含三个部分&#xff1a;模板&am…
阅读更多...
【ES6】—解构赋值

【ES6】—解构赋值

一、定义 解构赋值&#xff1a;解构赋值就是一种模式的匹配&#xff0c;只要等号两边的模式完全相同的&#xff0c;那么左边的变量就会被赋值对应右边的值 二、数组的解构赋值 PS&#xff1a;数组解构赋值时&#xff0c;是通过索引的唯一性赋值的 1. 一维数组解构赋值 (1)…
阅读更多...
大数据从入门到放弃——浅谈数据架构的前世今生

大数据从入门到放弃——浅谈数据架构的前世今生

文章目录 1. 背景2. 数据的定义及分类2.1 数据的定义2.2 数据的分类2.3 数据和信息的区别 3. 数据的作用4. 数据的那些美好时代4.1 人脑时代4.2 文件时代4.3 数据库时代4.3.1 大服务器时代4.3.2 读写分离时代4.4 数据库的分布式时代4.5 云端时代 5. 数据的未来 1. 背景 随着云时…
阅读更多...
真·VB.NET彻底释放Interop.Excel对象

真·VB.NET彻底释放Interop.Excel对象

使用 Microsoft.Office.Interop.Excel 虽然有速度慢的缺点&#xff1b;但是作为自带引用&#xff0c;兼容性最好&#xff0c;而且是COM对象模型也很熟悉(Excel里直接录个宏&#xff0c;很方便把VBA代码转成VB.NET)。所以处理几百上千条的小数据时还是很方便的。 而 Microsoft.…
阅读更多...
自学C#，要懂得善用MSDN

自学C#，要懂得善用MSDN

很多初学者学习编程&#xff0c;都会通过看别人写的教程、或者录制的视频&#xff0c;来学习。 这是一个非常好的途径&#xff0c;因为这个是非常高效的。 但是这样&#xff0c;存在两个问题&#xff1a; 1、教程不够全面&#xff1a;任何再好的教程&#xff0c;都无法囊括所…
阅读更多...
【C++_primary】类和对象 —— 类

【C++_primary】类和对象 —— 类

类 ~ ~ ~ 一、面向过程和面向对象初步认识a. 面向过程编程b. 面向对象编程例如&#xff1a;无人机送货系统1、面向过程编程方式2、面向对象编程方式 二、类的引入1、定义类的关键字2、栈的手动实现a. C语言实现栈b. C实现栈 三、类的定义类的两种定义方式&#xff1a; 四、类的…
阅读更多...
【Go】锁相关

【Go】锁相关

文章目录 Mutex锁mutex源码分析LockUnLock mutex两种运行模式mutex normal 正常模式自旋 mutex starvation 饥饿模式 锁的底层实现类型 RWMutexRWMutex 实现其他共享内存线程安全的方式 思考如何设计一个并发更高的锁&#xff1f; Mutex锁 mutex源码分析 Locker接口&#xff…
阅读更多...
【python知识点】锦集

【python知识点】锦集

【版权声明】未经博主同意&#xff0c;谢绝转载&#xff01;&#xff08;请尊重原创&#xff0c;博主保留追究权&#xff09; https://blog.csdn.net/m0_69908381/article/details/132368704 出自【进步*于辰的博客】 相关博文&#xff1a;【python细节、经验】锦集。 注&#…
阅读更多...
MyBatis入门配置及CURD实现

MyBatis入门配置及CURD实现

目录 一、MyBatis简介 1. 什么是 MyBatis ? 2. MyBatis的特性 3. 什么是持久层框架&#xff1f; 二、MyBatis环境配置 2.1 创建maven工程 2.2 导入相关pom依赖 2.3 导入jdbc配置文件 2.4 Mybatis相关插件安装 3.5 Mybatis-cfg.xml 核心配置 2.6 引入Log4j2日志文件…
阅读更多...
jinfo（Configuration Info for Java）Java配置信息工具

jinfo（Configuration Info for Java）Java配置信息工具

jinfo&#xff08;Configuration Info for Java&#xff09;Java配置信息工具 jinfo&#xff08;Configuration Info for Java&#xff09;的作用是实时查看和调整虚拟机各项参数 查看虚拟机启动时显式指定的参数列表 jps -v 查看未被显式指定的参数的系统默认值 1、找资料 …
阅读更多...
Vue2-配置脚手架、分析脚手架、render函数、ref属性、props配置项、mixin配置项、scoped样式、插件

Vue2-配置脚手架、分析脚手架、render函数、ref属性、props配置项、mixin配置项、scoped样式、插件

&#x1f954;:总有一段付出了没有回报的日子 是在扎根 更多Vue知识请点击——Vue.js VUE2-Day6 配置脚手架脚手架结构render函数vue.js与vue.runtime.xxx.js的区别引入render函数为什么要引入残缺的vue呢&#xff1f; 脚手架默认配置ref属性props配置项传递数据接收数据注意点…
阅读更多...
最新文章

Copyright @ 2022~2023