SpringBoot Kafka 整合使用及安裝教程
前提
假設(shè)你了解過 SpringBoot 和 Kafka。
1、SpringBoot
如果對 SpringBoot 不了解的話,建議去看看 DD 大佬 和 純潔的微笑 的系列博客。
2、Kafka
Kafka 的話可以看看我前兩天寫的博客 : Kafka 安裝及快速入門 學(xué)習(xí)的話自己開臺(tái)虛擬機(jī)自己手動(dòng)搭建環(huán)境吧,有條件的買服務(wù)器。
注意:一定要親自自己安裝實(shí)踐,接下來我們將這兩個(gè)進(jìn)行整合。
創(chuàng)建項(xiàng)目
項(xiàng)目整體架構(gòu):
使用 IDEA 創(chuàng)建 SpringBoot 項(xiàng)目,這個(gè)很簡單了,這里不做過多的講解。
1、pom 文件代碼如下:
<?xml version="1.0" encoding="UTF-8"?> <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>com.zhisheng</groupId> <artifactId>kafka-learning</artifactId> <version>0.0.1-SNAPSHOT</version> <packaging>jar</packaging> <name>kafka-learning</name> <description>Demo project for Spring Boot + kafka</description> <parent> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-parent</artifactId> <version>1.5.9.RELEASE</version> <relativePath/> <!-- lookup parent from repository --> </parent> <properties> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> <project.reporting.outputEncoding>UTF-8</project.reporting.outputEncoding> <java.version>1.8</java.version> </properties> <dependencies> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-web</artifactId> </dependency> <dependency> <groupId>org.projectlombok</groupId> <artifactId>lombok</artifactId> <optional>true</optional> </dependency> <dependency> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-starter-test</artifactId> <scope>test</scope> </dependency> <dependency> <groupId>org.springframework.kafka</groupId> <artifactId>spring-kafka</artifactId> <version>1.1.1.RELEASE</version> </dependency> <dependency> <groupId>com.google.code.gson</groupId> <artifactId>gson</artifactId> <version>2.8.2</version> </dependency> </dependencies> <build> <plugins> <plugin> <groupId>org.springframework.boot</groupId> <artifactId>spring-boot-maven-plugin</artifactId> </plugin> </plugins> </build> </project>
主要引入了 spring-kafka 、lombok 、 gson 依賴。
2、消息實(shí)體類 Message.java 如下:
@Datapublic class Message { private Long id; //id private String msg; //消息 private Date sendTime; //時(shí)間戳 }
3、消息發(fā)送類 KafkaSender.java
@Component @Slf4jpublic class KafkaSender { @Autowired private KafkaTemplate<String, String> kafkaTemplate; private Gson gson = new GsonBuilder().create(); //發(fā)送消息方法 public void send() { Message message = new Message(); message.setId(System .currentTimeMillis()); message.setMsg(UUID.randomUUID().toString()); message.setSendTime(new Date()) ; log.info("+++++++++++++++++++++ message = {}", gson.toJson(message)); kafkaTemplate.send ("zhisheng", gson.toJson(message)); } }
就這樣,發(fā)送消息代碼就實(shí)現(xiàn)了。
這里關(guān)鍵的代碼為 kafkaTemplate.send()
方法, zhisheng 是 Kafka 里的 topic ,這個(gè) topic 在 Java 程序中是不需要提前在 Kafka 中設(shè)置的,因?yàn)樗鼤?huì)在發(fā)送的時(shí)候自動(dòng)創(chuàng)建你設(shè)置的 topic, gson.toJson(message)
是消息內(nèi)容,這里暫時(shí)先說這么多了,不詳解了,后面有機(jī)會(huì)繼續(xù)把里面源碼解讀寫篇博客出來(因?yàn)橹型九龅娇樱献痈藥妆樵创a)。
4、消息接收類 KafkaReceiver.java
@Component @Slf4jpublic class KafkaReceiver { @KafkaListener(topics = {"zhisheng"}) public void listen(ConsumerRecord<?, ?> record) { Optional<?> kafkaMessage = Optional.ofNullable(record.value()); if (kafkaMessage.isPresent()) { Object message = kafkaMessage.get(); log.info("----------------- record =" + record); log.info("------------------ message =" + message); } } }
客戶端 consumer 接收消息特別簡單,直接用 @KafkaListener
注解即可,并在監(jiān)聽中設(shè)置監(jiān)聽的 topic , topics 是一個(gè)數(shù)組所以是可以綁定多個(gè)主題的,上面的代碼中修改為 @KafkaListener(topics={"zhisheng","tian"})
就可以同時(shí)監(jiān)聽兩個(gè) topic 的消息了。需要注意的是:這里的 topic 需要和消息發(fā)送類 KafkaSender.java 中設(shè)置的 topic 一致。
5、啟動(dòng)類 KafkaApplication.java
@SpringBootApplicationpublic class KafkaApplication { public static void main(String[] args) { ConfigurableApplicationContext context = SpringApplication.run(KafkaApplication.class, args); KafkaSender sender = context.getBean(KafkaSender.class); for (int i = 0; i < 3; i++) { //調(diào)用消息發(fā)送類中的消息發(fā)送方法 sender.send(); try { Thread.sleep(3000); } catch (InterruptedException e) { e.printStackTrace(); } } } }
6、配置文件 application.properties
#============== kafka ===================# 指定kafka 代理地址,可以多個(gè) spring.kafka.bootstrap-servers=192.168.153.135:9092 ##=============== provider ======================= #spring.kafka.producer.retries=0# 每次批量發(fā)送消息的數(shù)量 spring.kafka.producer.batch-size=16384spring.kafka.producer.buffer-memory=33554432 ## 指定消息key和消息體的編解碼方式 spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer spring.kafka.producer.value-serializer=org.apache.kafka.common.serialization.StringSerializer ##=============== consumer =======================# 指定默認(rèn)消費(fèi)者group id spring.kafka.consumer.group-id=test-consumer-group spring.kafka.consumer.auto-offset-reset=earliest spring.kafka.consumer.enable-auto-commit=true spring.kafka.consumer.auto-commit-interval=100 ## 指定消息key和消息體的編解碼方式 spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer spring.kafka.consumer.value-deserializer=org.apache.kafka.common.serialization.StringDeserializer
spring.kafka.bootstrap-servers
后面設(shè)置你安裝的 Kafka 的機(jī)器 IP 地址和端口號 9092。
如果你只是簡單整合下,其他的幾個(gè)默認(rèn)就好了。
Kafka 設(shè)置
在你安裝的 Kafka 目錄文件下:
啟動(dòng) zk
使用安裝包中的腳本啟動(dòng)單節(jié)點(diǎn) Zookeeper 實(shí)例:
bin/zookeeper-server-start.sh -daemon config/zookeeper.properties
啟動(dòng) Kafka 服務(wù)
使用 kafka-server-start.sh 啟動(dòng) kafka 服務(wù):
bin/kafka-server-start.sh config/server.properties
啟動(dòng)成功后!
千萬注意:記得將你的虛擬機(jī)或者服務(wù)器關(guān)閉防火墻或者開啟 Kafka 的端口 9092。
運(yùn)行
出現(xiàn)這就代表整合成功了!
--------------------------------------------------------------------------------
我們看下 Kafka 中的 topic 列表就
bin/kafka-topics.sh --list --zookeeper localhost:2181
就會(huì)發(fā)現(xiàn)剛才我們程序中的 zhisheng 已經(jīng)自己創(chuàng)建了。
總結(jié)
以上所述是小編給大家介紹的SpringBoot Kafka 整合使用及安裝教程,希望對大家有所幫助,如果大家有任何疑問請給我留言,小編會(huì)及時(shí)回復(fù)大家的。在此也非常感謝大家對腳本之家網(wǎng)站的支持!
如果你覺得本文對你有幫助,歡迎轉(zhuǎn)載,煩請注明出處,謝謝!
相關(guān)文章
shade解決mybatis包沖突問題及項(xiàng)目引用的方法
這篇文章主要介紹了shade解決mybatis包沖突問題及項(xiàng)目引用的方法,本文給大家介紹的非常詳細(xì),對大家的學(xué)習(xí)或工作具有一定的參考借鑒價(jià)值,需要的朋友可以參考下2020-08-08Apache Commons Math3探索之快速傅立葉變換代碼示例
這篇文章主要介紹了Apache Commons Math3探索之快速傅立葉變換代碼示例,具有一定參考價(jià)值,需要的朋友可以了解下。2017-10-10關(guān)于Lists.partition集合分組使用以及注意事項(xiàng)
這篇文章主要介紹了關(guān)于Lists.partition集合分組使用以及注意事項(xiàng),具有很好的參考價(jià)值,希望對大家有所幫助,如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2024-01-01Java多線程中ReentrantLock與Condition詳解
這篇文章主要介紹了Java多線程中ReentrantLock與Condition詳解,需要的朋友可以參考下2017-11-11SpringCloud如何利用Feign訪問外部http請求
這篇文章主要介紹了SpringCloud如何利用Feign訪問外部http請求,具有很好的參考價(jià)值,希望對大家有所幫助。如有錯(cuò)誤或未考慮完全的地方,望不吝賜教2022-03-03詳解Spring Boot + Mybatis 實(shí)現(xiàn)動(dòng)態(tài)數(shù)據(jù)源
這篇文章主要介紹了Spring Boot + Mybatis 實(shí)現(xiàn)動(dòng)態(tài)數(shù)據(jù)源,文中通過示例代碼介紹的非常詳細(xì),對大家的學(xué)習(xí)或者工作具有一定的參考學(xué)習(xí)價(jià)值,需要的朋友們下面隨著小編來一起學(xué)習(xí)學(xué)習(xí)吧2019-04-04