Springboot集成Kafka實(shí)現(xiàn)producer和consumer的示例代碼
本文介紹如何在springboot項(xiàng)目中集成kafka收發(fā)message。
Kafka是一種高吞吐量的分布式發(fā)布訂閱消息系統(tǒng)，有如下特性： 通過(guò)O(1)的磁盤(pán)數(shù)據(jù)結(jié)構(gòu)提供消息的持久化，這種結(jié)構(gòu)對(duì)于即使數(shù)以TB的消息存儲(chǔ)也能夠保持長(zhǎng)時(shí)間的穩(wěn)定性能。高吞吐量：即使是非常普通的硬件Kafka也可以支持每秒數(shù)百萬(wàn)的消息。支持通過(guò)Kafka服務(wù)器和消費(fèi)機(jī)集群來(lái)分區(qū)消息。支持Hadoop并行數(shù)據(jù)加載。
安裝Kafka
因?yàn)榘惭bkafka需要zookeeper的支持，所以Windows安裝時(shí)需要將zookeeper先安裝上，然后將kafka安裝好就可以了。 下面我給出Mac安裝的步驟以及需要注意的點(diǎn)吧，windows的配置除了所在位置不太一樣其他幾乎沒(méi)什么不同。
brew install kafka
對(duì)，就是這么簡(jiǎn)單，mac上一個(gè)命令就可以搞定了，這個(gè)安裝過(guò)程可能需要等一會(huì)兒，應(yīng)該是和網(wǎng)絡(luò)狀況有關(guān)系。安裝提示信息可能有錯(cuò)誤消息，如"Error: Could not link: /usr/local/share/doc/homebrew" 這個(gè)沒(méi)關(guān)系，自動(dòng)忽略掉了。 最終我們看到下面的樣子就成功咯。
==> Summary ðŸº/usr/local/Cellar/kafka/1.1.0: 157 files, 47.8MB
安裝的配置文件位置如下，根據(jù)自己的需要修改端口號(hào)什么的就可以了。
安裝的zoopeeper和kafka的位置 /usr/local/Cellar/
配置文件 /usr/local/etc/kafka/server.properties /usr/local/etc/kafka/zookeeper.properties
啟動(dòng)zookeeper
啟動(dòng)kafka
./bin/kafka-server-start /usr/local/etc/kafka/server.properties &
為kafka創(chuàng)建Topic,topic 名為test，可以配置成自己想要的名字，回頭再代碼中配置正確就可以了。
1、先解決依賴
springboot相關(guān)的依賴我們就不提了，和kafka相關(guān)的只依賴一個(gè)spring-kafka集成包
<dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> <version>1.1.1.RELEASE</version> </dependency>
這里我們先把配置文件展示一下
#============== kafka =================== kafka.consumer.zookeeper.connect=10.93.21.21:2181 kafka.consumer.servers=10.93.21.21:9092 kafka.consumer.enable.auto.commit=true kafka.consumer.session.timeout=6000 kafka.consumer.auto.commit.interval=100 kafka.consumer.auto.offset.reset=latest kafka.consumer.topic=test kafka.consumer.group.id=test kafka.consumer.concurrency=10 kafka.producer.servers=10.93.21.21:9092 kafka.producer.retries=0 kafka.producer.batch.size=4096 kafka.producer.linger=1 kafka.producer.buffer.memory=40960
2、Configuration：Kafka producer
1）通過(guò)@Configuration、@EnableKafka，聲明Config并且打開(kāi)KafkaTemplate能力。
2）通過(guò)@Value注入application.properties配置文件中的kafka配置。
3）生成bean，@Bean
package com.kangaroo.sentinel.collect.configuration;
import java.util.HashMap;
import java.util.Map;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringSerializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
@Configuration
@EnableKafka
public class KafkaProducerConfig {
@Value("${kafka.producer.servers}")
private String servers;
@Value("${kafka.producer.retries}")
private int retries;
@Value("${kafka.producer.batch.size}")
private int batchSize;
@Value("${kafka.producer.linger}")
private int linger;
@Value("${kafka.producer.buffer.memory}")
private int bufferMemory;
public Map<String, Object> producerConfigs() {
Map<String, Object> props = new HashMap<>();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
props.put(ProducerConfig.RETRIES_CONFIG, retries);
props.put(ProducerConfig.BATCH_SIZE_CONFIG, batchSize);
props.put(ProducerConfig.LINGER_MS_CONFIG, linger);
props.put(ProducerConfig.BUFFER_MEMORY_CONFIG, bufferMemory);
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
return props;
}
public ProducerFactory<String, String> producerFactory() {
return new DefaultKafkaProducerFactory<>(producerConfigs());
}
@Bean
public KafkaTemplate<String, String> kafkaTemplate() {
return new KafkaTemplate<String, String>(producerFactory());
}
}
實(shí)驗(yàn)我們的producer，寫(xiě)一個(gè)Controller。想topic=test，key=key，發(fā)送消息message
package com.kangaroo.sentinel.collect.controller;
import com.kangaroo.sentinel.common.response.Response;
import com.kangaroo.sentinel.common.response.ResultCode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.*;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;
@RestController
@RequestMapping("/kafka")
public class CollectController {
protected final Logger logger = LoggerFactory.getLogger(this.getClass());
@Autowired
private KafkaTemplate kafkaTemplate;
@RequestMapping(value = "/send", method = RequestMethod.GET)
public Response sendKafka(HttpServletRequest request, HttpServletResponse response) {
try {
String message = request.getParameter("message");
logger.info("kafka的消息={}"， message);
kafkaTemplate.send("test", "key", message);
logger.info("發(fā)送kafka成功.");
return new Response(ResultCode.SUCCESS, "發(fā)送kafka成功", null);
} catch (Exception e) {
logger.error("發(fā)送kafka失敗", e);
return new Response(ResultCode.EXCEPTION, "發(fā)送kafka失敗", null);
}
}
}
3、configuration：kafka consumer
1）通過(guò)@Configuration、@EnableKafka，聲明Config并且打開(kāi)KafkaTemplate能力。
2）通過(guò)@Value注入application.properties配置文件中的kafka配置。
3）生成bean，@Bean
package com.kangaroo.sentinel.collect.configuration;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.config.KafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ConcurrentMessageListenerContainer;
import java.util.HashMap;
import java.util.Map;
@Configuration
@EnableKafka
public class KafkaConsumerConfig {
@Value("${kafka.consumer.servers}")
private String servers;
@Value("${kafka.consumer.enable.auto.commit}")
private boolean enableAutoCommit;
@Value("${kafka.consumer.session.timeout}")
private String sessionTimeout;
@Value("${kafka.consumer.auto.commit.interval}")
private String autoCommitInterval;
@Value("${kafka.consumer.group.id}")
private String groupId;
@Value("${kafka.consumer.auto.offset.reset}")
private String autoOffsetReset;
@Value("${kafka.consumer.concurrency}")
private int concurrency;
@Bean
public KafkaListenerContainerFactory<ConcurrentMessageListenerContainer<String, String>> kafkaListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<String, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
factory.setConsumerFactory(consumerFactory());
factory.setConcurrency(concurrency);
factory.getContainerProperties().setPollTimeout(1500);
return factory;
}
public ConsumerFactory<String, String> consumerFactory() {
return new DefaultKafkaConsumerFactory<>(consumerConfigs());
}
public Map<String, Object> consumerConfigs() {
Map<String, Object> propsMap = new HashMap<>();
propsMap.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, servers);
propsMap.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, enableAutoCommit);
propsMap.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, autoCommitInterval);
propsMap.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, sessionTimeout);
propsMap.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
propsMap.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
propsMap.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
propsMap.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);
return propsMap;
}
@Bean
public Listener listener() {
return new Listener();
}
}
new Listener()生成一個(gè)bean用來(lái)處理從kafka讀取的數(shù)據(jù)。Listener簡(jiǎn)單的實(shí)現(xiàn)demo如下：只是簡(jiǎn)單的讀取并打印key和message值
@KafkaListener中topics屬性用于指定kafka topic名稱，topic名稱由消息生產(chǎn)者指定，也就是由kafkaTemplate在發(fā)送消息時(shí)指定。
package com.kangaroo.sentinel.collect.configuration;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
public class Listener {
protected final Logger logger = LoggerFactory.getLogger(this.getClass());
@KafkaListener(topics = {"test"})
public void listen(ConsumerRecord<?, ?> record) {
logger.info("kafka的key: " + record.key());
logger.info("kafka的value: " + record.value().toString());
}
}
tips:
1）我沒(méi)有介紹如何安裝配置kafka，配置kafka時(shí)最好用完全bind網(wǎng)絡(luò)ip的方式，而不是localhost或者127.0.0.1
2）最好不要使用kafka自帶的zookeeper部署kafka，可能導(dǎo)致訪問(wèn)不通。
3）理論上consumer讀取kafka應(yīng)該是通過(guò)zookeeper，但是這里我們用的是kafkaserver的地址，為什么沒(méi)有深究。
4）定義監(jiān)聽(tīng)消息配置時(shí)，GROUP_ID_CONFIG配置項(xiàng)的值用于指定消費(fèi)者組的名稱，如果同組中存在多個(gè)監(jiān)聽(tīng)器對(duì)象則只有一個(gè)監(jiān)聽(tīng)器對(duì)象能收到消息。
以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助，也希望大家多多支持腳本之家。
- 如何使用SpringBoot集成Kafka實(shí)現(xiàn)用戶數(shù)據(jù)變更后發(fā)送消息
- SpringBoot3集成Kafka的方法詳解
- SpringBoot集成Kafka 配置工具類的詳細(xì)代碼
- springboot集成kafka消費(fèi)手動(dòng)啟動(dòng)停止操作
- Springboot集成kafka高級(jí)應(yīng)用實(shí)戰(zhàn)分享
- Springboot 2.x集成kafka 2.2.0的示例代碼
- SpringBoot集成kafka全面實(shí)戰(zhàn)記錄
- SpringBoot集成Kafka的步驟
- SpringBoot集成Kafka的實(shí)現(xiàn)示例
相關(guān)文章
SpringCloud-Nacos服務(wù)注冊(cè)與發(fā)現(xiàn)方式
這篇文章主要介紹了SpringCloud-Nacos服務(wù)注冊(cè)與發(fā)現(xiàn)方式,具有很好的參考價(jià)值,希望對(duì)大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2024-07-07
今天小編就為大家分享一篇關(guān)于Java類加載初始化的過(guò)程及順序，小編覺(jué)得內(nèi)容挺不錯(cuò)的，現(xiàn)在分享給大家，具有很好的參考價(jià)值，需要的朋友一起跟隨小編來(lái)看看吧2018-12-12
SpringBoot 3.0 新特性內(nèi)置聲明式HTTP客戶端實(shí)例詳解
聲明式 http 客戶端主旨是使得編寫(xiě) java http 客戶端更容易，為了貫徹這個(gè)理念，采用了通過(guò)處理注解來(lái)自動(dòng)生成請(qǐng)求的方式，本文給大家詳解介紹SpringBoot 聲明式HTTP客戶端相關(guān)知識(shí)，感興趣的朋友跟隨小編一起看看吧2022-12-12
Java concurrency之LockSupport_動(dòng)力節(jié)點(diǎn)Java學(xué)院整理
這篇文章主要為大家詳細(xì)介紹了Java concurrency之LockSupport的相關(guān)資料，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2017-06-06
學(xué)習(xí)Java的static與final關(guān)鍵字
本篇文章給大家詳細(xì)分析了Java的static與final關(guān)鍵字知識(shí)點(diǎn)以及相關(guān)代碼分享，有需要的讀者跟著學(xué)習(xí)下吧。2018-03-03
MySQL查詢字段實(shí)現(xiàn)字符串分割split功能的示例代碼
本文主要介紹了MySQL查詢字段實(shí)現(xiàn)字符串分割split功能的示例代碼，文中通過(guò)示例代碼介紹的非常詳細(xì)，具有一定的參考價(jià)值，感興趣的小伙伴們可以參考一下2022-01-01
MyBatis源碼解析之Transaction事務(wù)模塊
這篇文章主要介紹了MyBatis源碼解析之Transaction事務(wù)模塊，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。如有錯(cuò)誤或未考慮完全的地方，望不吝賜教2021-10-10
這篇文章主要為大家介紹了kafka?rebalance負(fù)載均衡的深入理解，有需要的朋友可以借鑒參考下，希望能夠有所幫助，祝大家多多進(jìn)步，早日升職加薪2023-03-03
這篇文章主要給大家介紹了關(guān)于XML操作類庫(kù)XStream使用的相關(guān)資料,需要的朋友可以參考下2023-11-11
這篇文章主要介紹了Java 類動(dòng)態(tài)添加屬性字段的操作，具有很好的參考價(jià)值，希望對(duì)大家有所幫助。一起跟隨小編過(guò)來(lái)看看吧2021-02-02