【基于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

【基于XML方式】Spring整合Kafka相关推荐

Spring-注入方式(基于xml方式)
1.基于xml方式创建对象 <bean id="user" class="com.at.spring5.User ...
Spring整合Kafka
文章目录 Spring整合Kafka 一.引入依赖二.配置kafka 三.测试代码--如何用kafka 3.1 KafkaTests 3.2 测试结果 Spring整合Kafka 一.引入依赖 &l ...
Spring IOC容器-Bean管理——基于XML方式
Spring IOC容器-Bean管理--基于XML(续集) 1.IOC 操作 Bean 管理(FactoryBean) 1).Spring 有两种类型 bean,一种普通 bean,另外一种工厂 ...
spring入门第二讲 bean的生命周期装配方式 Spring整合Junit
bean的生命周期实体类 //初始化 public void init(){System.out.println("--初始化--"); }//销毁 public void de ...
【基于注解方式】Spring整合Kafka
文章目录 1. 添加Maven依赖 2. 配置与参数分离 3. 工具类度内容 4. Producer 消息生产者配置 5. Consumer 消息消费者配置 6. 使用注解监听消息 7. 请求测试 8 ...
spring的IOC容器Bean管理（基于xml方式）
==目的:==降低耦合度 ==技术:==xml解析.工厂模式.反射 ioc容器的实现 spring框架提供了两种实现方式(两个接口) BeanFactory接口 ioc容器的基本实现,是spring框 ...
Spring的事务控制-基于xml方式
介绍:该程序模拟了转账操作,即Jone减少500元,tom增加500元 1.导入坐标 <dependency><groupId>junit</groupId>< ...
spring 基于xml方式配置aop
目录什么是aop 模拟aop 配置什么是aop 什么是aop 作用在程序运行期间,在不修改源码的情况下对方法进行功能增强优势减少重复代码提高开发效率并且便于开发 2.aop关键概念模拟 ...
Spring学习笔记之一----基于XML的Spring IOC配置
1. 在spring配置文件中,如果对一个property进行直接赋值,可使用<value>元素,spring负责将值转化为property指定的类型:也可以直接在property元素上使 ...

【基于XML方式】Spring整合Kafka

文章目录

1.引入Maven依赖

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

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

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

5. kafka消息监听类

6. Controller的实现

7. 测试

8. 项目链接：

【基于XML方式】Spring整合Kafka相关推荐

最新文章

热门文章