【基于XML方式】Spring整合Kafka

文章目录

      • 1.引入Maven依赖
      • 2. kafka.properties文件中配置内容
      • 3. 发送端 producer-kafka.xml 配置
      • 4. 接收端 consumer-kafka.xml 配置
      • 5. kafka消息监听类
      • 6. Controller的实现
      • 7. 测试
      • 8. 项目链接:

1.引入Maven依赖

<!-- 添加spring-kafka支持 -->
<dependency><groupId>org.springframework.kafka</groupId><artifactId>spring-kafka</artifactId><version>2.2.2.RELEASE</version>
</dependency>

2. kafka.properties文件中配置内容

################## kafkaListener Producer 发送端配置 ##################
# brokers 集群
kafka.producer.bootstrap.servers = 192.168.204.201:9092,192.168.204.202:9092,192.168.204.203:9092#发送端 id
kafka.producer.client.id = producerDemo#发送端确认模式
kafka.producer.acks = -1#发送失败重试次数
kafka.producer.retries = 3#批处理条数,当多个记录被发送至统一分区时,producer对于同一个分区来说,会按照 batch.size 的大小进行统一收集,批量发送
kafka.producer.batch.size = 4096#与 batch.size 配合使用。延迟统一收集,产生聚合,然后批量发送至broker
kafka.producer.linger.ms = 10# 33554432 即32MB的批处理缓冲区
kafka.producer.buffer.memory = 40960#默认 topic
kafka.producer.defaultTopic = testTopic#key 序列化
kafka.producer.key.serializer = org.apache.kafka.common.serialization.StringSerializer#value 序列化
kafka.producer.value.serializer = org.apache.kafka.common.serialization.StringSerializer################## kafkaListener Consumer 消费端配置 ###################消费端 brokers 集群
kafka.consumer.bootstrap.servers = 192.168.204.201:9092,192.168.204.202:9092,192.168.204.203:9092#消费者 group.id 组ID
kafka.consumer.group.id = test-group#消费者消费消息后,进行自动提交(offsets偏移量)
kafka.consumer.enable.auto.commit = true#自动提交的频率(与 enable.auto.commit = true 属性配合使用)
kafka.consumer.auto.commit.interval.ms = 1000#新的groupid,是否从头开始消费
kafka.consumer.auto.offset.reset = earliest#在使用kafka组管理时,发送心跳机制,用于检测消费者故障的超时
#kafka.consumer.session.timeout.ms = 1000#key 反序列化
kafka.consumer.key.deserializer = org.apache.kafka.common.serialization.StringDeserializer#value 反序列化
kafka.consumer.value.deserializer = org.apache.kafka.common.serialization.StringDeserializer#消费端消费的topic
kafka.consumer.topic = testTopic

3. 发送端 producer-kafka.xml 配置

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://www.springframework.org/schema/beanshttp://www.springframework.org/schema/beans/spring-beans.xsd"><!-- 1.定义producer的参数 --><bean id="producerProperties" class="java.util.HashMap"><constructor-arg><map><entry key="bootstrap.servers" value="${kafka.producer.bootstrap.servers}"/><entry key="client.id" value="${kafka.producer.client.id}"/><entry key="retries" value="${kafka.producer.retries}"/><entry key="batch.size" value="${kafka.producer.batch.size}"/><entry key="linger.ms" value="${kafka.producer.linger.ms}"/><entry key="buffer.memory" value="${kafka.producer.buffer.memory}"/><entry key="acks" value="${kafka.producer.acks}"/><entry key="key.serializer" value="${kafka.producer.key.serializer}"/><entry key="value.serializer" value="${kafka.producer.value.serializer}"/></map></constructor-arg></bean><!-- 2.创建 kafkaTemplate 需要使用的 producerFactory Bean --><bean id="producerFactory" class="org.springframework.kafka.core.DefaultKafkaProducerFactory"><constructor-arg ref="producerProperties"/></bean><!-- 3.创建 kafkaTemplate Bean,使用的时候只需要注入这个bean,即可使用kafkaTemplate的 send 消息方法 --><bean id="kafkaTemplate" class="org.springframework.kafka.core.KafkaTemplate"><constructor-arg ref ="producerFactory"/><constructor-arg name="autoFlush" value="true"/><property name="defaultTopic" value="${kafka.producer.defaultTopic}"/></bean></beans>

4. 接收端 consumer-kafka.xml 配置

<?xml version="1.0" encoding="UTF-8"?>
<beans xmlns="http://www.springframework.org/schema/beans"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://www.springframework.org/schema/beanshttp://www.springframework.org/schema/beans/spring-beans.xsd"><!-- 1.定义 consumer 的参数 --><bean id="consumerProperties" class="java.util.HashMap"><constructor-arg><map><entry key="bootstrap.servers" value="${kafka.consumer.bootstrap.servers}"/><entry key="group.id" value="${kafka.consumer.group.id}"/><entry key="enable.auto.commit" value="${kafka.consumer.enable.auto.commit}"/><entry key="auto.commit.interval.ms" value="${kafka.consumer.auto.commit.interval.ms}"/><entry key="key.deserializer" value="${kafka.consumer.key.deserializer}"/><entry key="value.deserializer" value="${kafka.consumer.value.deserializer}"/><entry key="auto.offset.reset" value="${kafka.consumer.auto.offset.reset}"/></map></constructor-arg></bean><!-- 2.创建 consumerFactory bean --><bean id="consumerFactory" class="org.springframework.kafka.core.DefaultKafkaConsumerFactory"><constructor-arg ref="consumerProperties"/></bean><!-- 3.配置消费端 Kafka 监听实现类(自己实现) --><bean id="kafkaConsumerListener" class="com.demo.kafkaListener.KafkaConsumerListener"/><!-- 4.消费者容器配置 --><bean id="containerProperties" class="org.springframework.kafka.listener.ContainerProperties"><constructor-arg name="topics"><list><value>${kafka.consumer.topic}</value></list></constructor-arg><property name="messageListener" ref="kafkaConsumerListener"/></bean><!-- 5.消费者并发消息监听容器 --><bean id="messageListenerContainer" class="org.springframework.kafka.listener.KafkaMessageListenerContainer"><constructor-arg ref="consumerFactory"/><constructor-arg ref="containerProperties"/></bean></beans>

5. kafka消息监听类

/*** 监听类的实现*/
public class KafkaConsumerListener implements MessageListener<String,String> {public void onMessage(ConsumerRecord<String, String> stringStringConsumerRecord) {System.out.println("========消费端收到消息========");System.out.println(stringStringConsumerRecord.value());//根据不同主题,消费if("主题1".equals(stringStringConsumerRecord.topic())){//逻辑1}else if("主题2".equals(stringStringConsumerRecord.topic())){//逻辑2}}}

6. Controller的实现

@Controller
@RequestMapping("kafka")
public class KafkaController {@AutowiredKafkaTemplate kafkaTemplate;/*** 消息发送*/@RequestMapping("producer")public String producer(){kafkaTemplate.send("testTopic","producer发送消息");return "success";}}

7. 测试

通过 http://localhsot:8080/kafka/producer
执行消息发送操作,在监听类 KafkaConsumerListener 中完成对消息的消费。

8. 项目链接:

https://github.com/gb-heima/spring-kafka

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

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

相关文章

使用CLI进行高速跨域日志复制、历史数据重新索引与数仓投递

背景 使用日志服务是不是常常遇到如下烦恼&#xff1f; 开启了字段索引却无法对历史日志起作用&#xff0c;而手动重建索引又很困难怎么办&#xff1f;需要迁移数据&#xff0c;复制数据到其他区域logstore&#xff0c;写代码实现大并发复制又很复杂怎么办&#xff1f;投递日…

java map大小_Java中HashMap的size()方法: HashMap.size() - Break易站

Java中的HashMapHashMap类的java.util.HashMap.size()方法用于获取哈希映射HashMap中映射的大小&#xff0c;该映射引用Map中键值对或映射的数量。句法&#xff1a;Hash_Map.size()参数&#xff1a;该方法不接受任何参数。返回值&#xff1a;该方法返回映射的大小&#xff0c;这…

gitee提交代码_git 版本控制,github和gitee

3.4 git 版本控制# 查看本地状态git status # 修改README.txt &#xff0c;添加一行,保存&#xff0c;添加到暂存区git add README.txt # 查看当前仓库某一个文件的版本git diff README.txt# 提交到本地仓库git commit README.txt -m second Commit# 查看当前仓库某一个文件版本…

IT 拉呱室 | 论我遇到的最刺激的bug【长期福利站】

戳蓝字“CSDN云计算”关注我们哦&#xff01;套路的最高境界是没有套路&#xff1a;这是一个只送福利的长期互动栏目【如果你们让我尴尬&#xff0c;我就哭给你们看】。书籍、键盘、鼠标、鼠标垫、CSDN 20周年纪念T恤、技术沙龙门票、线上公开课名额……你敢一直在&#xff0c;…

“百变”Redis带你见识不同场景下的产品技术架构

2018飞天技术汇24期-云数据库Redis产品发布会&#xff0c;由阿里云数据库技术组技术专家王欢、怀听、梁盼分别带来以“Redis全球多活产品”、“Redis混合存储产品”、“Redis多线程性能增强版”为题的演讲。本文对Redis进行了简单的介绍&#xff0c;进而针对不同的应用场景研制…

【基于注解方式】Spring整合Kafka

文章目录1. 添加Maven依赖2. 配置与参数分离3. 工具类度内容4. Producer 消息生产者配置5. Consumer 消息消费者配置6. 使用注解监听消息7. 请求测试8. 测试结果1. 添加Maven依赖 <!-- 添加spring-kafka支持 --> <dependency><groupId>org.springframework.…

阿里云PyODPS 0.7.18发布,针对聚合函数进行优化同时新增对Python 3.7支持

近日&#xff0c;阿里云发布PyODPS 0.7.18&#xff0c;主要是针对聚合函数进行优化同时新增对Python 3.7支持。 PyODPS是MaxCompute的Python版本的SDK&#xff0c;SDK的意思非常广泛&#xff0c;辅助开发某一类软件的相关文档、范例和工具的集合都可以叫做“SDK”。 PyODPS在这…

centos安装rabbitmq_【SpringBoot MQ系列教程】RabbitMq 初体验

SpringBoot 系列教程之 RabbitMq 初体验​mp.weixin.qq.commq 在异步解耦削峰的优势非常突出&#xff0c;现在很多的项目都会用到&#xff0c;掌握 mq 的知识点&#xff0c;了解如何顺畅的使用 mq&#xff0c;可以说是一个必备的职业技能点了接下来我们进入 rabbitmq 的学习过程…

java rpc与webservice_RPC体系,RPC和WebService的区别详解

RPC和WebService的关系RPC(Remote Procedure Call)— 远程过程调用&#xff0c;是一个很大的概念, 它是一种通过网络从远程计算机程序上跨语言跨平台的请求服务&#xff0c;rpc能省略部分接口代码的开发&#xff0c;可以跨机器之间访问对象(java rmi)&#xff0c;可以有更方便的…

免费公测中-GPU数据库SQream DB正式上线云市场

业内领先的GPU 数据库服务SQream DB在阿里云云市场正式开启免费公测&#xff01;SQream DB是一款由阿里战略投资的以色列SQream公司提供&#xff0c;能够支撑海量数据高速分析的业内领先的GPU数据库。通过将计算密集型操作卸载到GPU上&#xff0c;与业界的解决方案相比&#xf…

唏嘘!2019榜单出炉:铁打的Python连续3年第一,它居然跌出前十?

IEEE Spectrum2019年度编程语言排行榜最近刚刚出炉&#xff0c;Python不出意外的又拿了个第一&#xff0c;但是意料之外的是&#xff0c;曾经大火的PHP&#xff0c;居然跌出了前十&#xff01;PHP曾被大家称为“世界上最好的编程语言”&#xff0c;去年排名第六&#xff0c;前年…

centos7无法使用epel的解决方法

使用如下代码在centos7安装epel源&#xff0c;却无法使用。 yum -y install epel-release网络没问题&#xff0c;可以ping通epel源的地址&#xff0c;但是就是连不上repo&#xff0c;报错。 解决方法&#xff1a;编辑/etc/yum.repos.d/epel.repo&#xff0c;将epel配置信息中…

阿里云ECS家族再添新成员,推出密集计算型实例规格族ic5

去年&#xff0c;阿里云正式发布云服务器ECS企业级产品家族&#xff0c;推出面向173种企业应用场景的19款实例。适合在复杂的企业计算环境下&#xff0c;满足对于高性能、高可靠的计算需求。 时隔近一年&#xff0c;回看ECS企业级产品家族已经发展到30款实例&#xff0c;近日再…

findbugs插件_Intellij静态代码扫描插件SpotBugs

最近要做Java静态扫描的部分工作&#xff0c;之前是在Jenkins上使用findbugs插件完成的&#xff0c;但是由于现在Jenkins权限收回和Java代码权限的放开(我也搞不懂这两者的关联性)&#xff0c;目前打算在本地完成静态代码扫描工作。选来选取还是选择在Intellij中用插件来完成&a…

一张图看懂智联车管理云平台

智联车管理云平台&#xff08;IoV Command Center&#xff0c;简称IoV CC&#xff09;是阿里云面向智联车领域&#xff0c;专门推出的车辆全生命周期云端管理平台&#xff0c;旨在赋能车厂转型出行服务商&#xff0c;提高运营效率、降低自建成本。 传统模式下&#xff0c;车辆…

刨根问底 | 红遍全网的SD-WAN,到底是个啥?

戳蓝字“CSDN云计算”关注我们哦&#xff01;作者 | 小枣君责编 | 阿秃作为一个热门概念&#xff0c;SD-WAN近年以来频繁地出现在我们的视野当中。很多人说&#xff0c;它是未来最具发展潜力的通信技术之一&#xff0c;极具商业价值。行业里的老牌通信设备商和运营商对它一致看…

centos7安装rabbitmq简单方式

安装rabbitmq前要准备的基础环境 yum install build-essential openssl openssl-devel unixODBC unixODBC-devel make gcc gcc-c kernel-devel m4 ncurses-devel tk tc xz tcp_wrappers需下载的安装文件如下 ## erlang wget www.rabbitmq.com/releases/erlang/erlang-18.3-1.…

java socket数据传输_Java Socket编程(一) Socket传输模式

Java Socket编程(一) Socket传输模式文章来源&#xff1a;ASPCN 作者&#xff1a;孙雯Socket传输模式Sockets有两种主要的操作方式:面向连接的和无连接的.面向连接的sockets操作就像一部电话,他们必须建立一个连接和一人呼叫.所有的事情在到达时的顺序与它们出发时的顺序时一样…

车联网上云最佳实践(一)

一、车联网行业特性讲解 最近两年车联网发展受到政府部门、科研院以及各大互联网巨头的广泛关注和积极推动。从应用来看&#xff0c;主要包括两种模式&#xff1a;一是前装模式&#xff08;即车辆出厂前安装&#xff09;&#xff0c;是乘用车厂主导或者与有相关能力的公司合作&…

python3 网站状态监控_基于python3监控服务器状态进行邮件报警

在正式的生产环境中&#xff0c;我们常常会需要监控服务器的状态&#xff0c;以保证公司整个业务的正常运转&#xff0c;常常我们会用到像nagios、zabbix这类工具进行实时监控&#xff0c;那么用python我们怎么进行监控呢&#xff1f;这里我们利用了python3调用psutil和yagmail…