前提
創(chuàng)新互聯(lián)堅(jiān)信:善待客戶,將會(huì)成為終身客戶。我們能堅(jiān)持多年,是因?yàn)槲覀円恢笨芍档眯刨?。我們從不忽悠初訪客戶,我們用心做好本職工作,不忘初心,方得始終。10多年網(wǎng)站建設(shè)經(jīng)驗(yàn)創(chuàng)新互聯(lián)是成都老牌網(wǎng)站營(yíng)銷服務(wù)商,為您提供成都網(wǎng)站制作、做網(wǎng)站、外貿(mào)營(yíng)銷網(wǎng)站建設(shè)、網(wǎng)站設(shè)計(jì)、H5高端網(wǎng)站建設(shè)、網(wǎng)站制作、品牌網(wǎng)站設(shè)計(jì)、重慶小程序開發(fā)服務(wù),給眾多知名企業(yè)提供過好品質(zhì)的建站服務(wù)。
假設(shè)你了解過 SpringBoot 和 Kafka。
1、SpringBoot
如果對(duì) 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è)很簡(jiǎn)單了,這里不做過多的講解。
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 接收消息特別簡(jiǎn)單,直接用 @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 地址和端口號(hào) 9092。
如果你只是簡(jiǎn)單整合下,其他的幾個(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 整合使用及安裝教程,希望對(duì)大家有所幫助,如果大家有任何疑問請(qǐng)給我留言,小編會(huì)及時(shí)回復(fù)大家的。在此也非常感謝大家對(duì)創(chuàng)新互聯(lián)網(wǎng)站的支持!
如果你覺得本文對(duì)你有幫助,歡迎轉(zhuǎn)載,煩請(qǐng)注明出處,謝謝!
分享名稱:SpringBootKafka整合使用及安裝教程
文章URL:http://bm7419.com/article22/ijhojc.html
成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián),為您提供品牌網(wǎng)站設(shè)計(jì)、移動(dòng)網(wǎng)站建設(shè)、網(wǎng)站制作、網(wǎng)站策劃、網(wǎng)站建設(shè)、網(wǎng)站營(yíng)銷
聲明:本網(wǎng)站發(fā)布的內(nèi)容(圖片、視頻和文字)以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主,如果涉及侵權(quán)請(qǐng)盡快告知,我們將會(huì)在第一時(shí)間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng),如需處理請(qǐng)聯(lián)系客服。電話:028-86922220;郵箱:631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載,或轉(zhuǎn)載時(shí)需注明來源: 創(chuàng)新互聯(lián)