當前位置：首頁 > 编程语言 > asp.net >内容正文

asp.net

【基于XML方式】Spring整合Kafka

發布時間：2024/9/27 asp.net 38 豆豆

生活随笔收集整理的這篇文章主要介紹了【基于XML方式】Spring整合Kafka 小編覺得挺不錯的,現在分享給大家,幫大家做個參考.

文章目錄

- - 1.引入Maven依賴
  - 2. kafka.properties文件中配置內容
  - 3. 發送端 producer-kafka.xml 配置
  - 4. 接收端 consumer-kafka.xml 配置
  - 5. kafka消息監聽類
  - 6. Controller的實現
  - 7. 測試
  - 8. 項目鏈接：

1.引入Maven依賴

<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"><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><bean id="producerFactory" class="org.springframework.kafka.core.DefaultKafkaProducerFactory"><constructor-arg ref="producerProperties"/></bean><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"><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><bean id="consumerFactory" class="org.springframework.kafka.core.DefaultKafkaConsumerFactory"><constructor-arg ref="consumerProperties"/></bean><bean id="kafkaConsumerListener" class="com.demo.kafkaListener.KafkaConsumerListener"/><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><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的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。

上一篇：端口占用8080
下一篇： Sublime Test3 添加一键格式