這篇文章主要介紹在Spring Boot應(yīng)用程序中如何使用Apache Kafka,文中介紹的非常詳細(xì),具有一定的參考價值,感興趣的小伙伴們一定要看完!
創(chuàng)新互聯(lián)公司2013年成立,是專業(yè)互聯(lián)網(wǎng)技術(shù)服務(wù)公司,擁有項(xiàng)目成都網(wǎng)站建設(shè)、成都做網(wǎng)站網(wǎng)站策劃,項(xiàng)目實(shí)施與項(xiàng)目整合能力。我們以讓每一個夢想脫穎而出為使命,1280元襄州做網(wǎng)站,已為上家服務(wù),為襄州各地企業(yè)和個人服務(wù),聯(lián)系電話:18980820575
第1步:生成我們的項(xiàng)目: Spring Initializr 來生成我們的項(xiàng)目。我們的項(xiàng)目將提供Spring MVC / Web支持和Apache Kafka支持。
第2步:發(fā)布/讀取Kafka主題中的消息:
<b>public</b> <b>class</b> User { <b>private</b> String name; <b>private</b> <b>int</b> age; <b>public</b> User(String name, <b>int</b> age) { <b>this</b>.name = name; <b>this</b>.age = age; } }
第3步:通過application.yml
配置文件配置Kafka:
我們需要創(chuàng)建配置文件。我們需要以某種方式配置我們的Kafka生產(chǎn)者和消費(fèi)者,以便能夠發(fā)布和讀取與主題相關(guān)的消息。相比建立一個使用@Configuration
標(biāo)注的Java類,我們可以直接使用配置文件application.properties或application.yml。Spring Boot讓我們避免像過去一樣編寫的所有樣板代碼,同時為我們提供了更加智能的配置應(yīng)用程序的方法,如下所示:
server: port: 9000 spring: kafka: consumer: bootstrap: localhost:9092 group-id: group_id auto-offset-reset: earliest key-deserializer: org.apache.kafka.common.serialization.StringDeserializer value-deserializer: org.apache.kafka.common.serialization.StringDeserializer producer: bootstrap: localhost:9092 key-serializer: org.apache.kafka.common.serialization.StringSerializer value-serializer: org.apache.kafka.common.serialization.StringSerializer
第4步:創(chuàng)建一個生產(chǎn)者,創(chuàng)建生產(chǎn)者會將我們的消息寫入該主題。
<b>public</b> <b>class</b> Producer { <b>private</b> <b>static</b> <b>final</b> Logger logger = LoggerFactory.getLogger(Producer.<b>class</b>); <b>private</b> <b>static</b> <b>final</b> String TOPIC = <font>"users"</font><font>; @Autowired <b>private</b> KafkaTemplate<String, String> kafkaTemplate; <b>public</b> <b>void</b> sendMessage(String message) { logger.info(String.format(</font><font>"#### -> Producing message -> %s"</font><font>, message)); <b>this</b>.kafkaTemplate.send(TOPIC, message); } } </font>
自動連接autowire
到 KafkaTemplate
,使用它將消息發(fā)布到主題 - 這就是消息的生產(chǎn)者!
第5步:創(chuàng)建一個消費(fèi)者,消費(fèi)者是負(fù)責(zé)根據(jù)您自己的業(yè)務(wù)邏輯的需求閱讀處理消息的消息的服務(wù)。要進(jìn)行設(shè)置,請輸入以下內(nèi)容:
@Service <b>public</b> <b>class</b> Consumer { <b>private</b> <b>final</b> Logger logger = LoggerFactory.getLogger(Producer.<b>class</b>); @KafkaListener(topics = <font>"users"</font><font>, groupId = </font><font>"group_id"</font><font>) <b>public</b> <b>void</b> consume(String message) throws IOException { logger.info(String.format(</font><font>"#### -> Consumed message -> %s"</font><font>, message)); } } </font>
在這里,我們告訴我們的方法void consume(String message)
訂閱用戶的主題,并將每條消息發(fā)送到應(yīng)用程序日志。在您的實(shí)際應(yīng)用程序中,您可以按照業(yè)務(wù)需要的方式處理消息。
第6步:創(chuàng)建REST控制器,們已經(jīng)擁有了能夠消費(fèi)Kafka消息所需的全部內(nèi)容。
為了充分展示我們創(chuàng)建的所有內(nèi)容的工作原理,我們需要創(chuàng)建一個具有單一端點(diǎn)的控制器。消息將發(fā)布到此端點(diǎn),然后由我們的生產(chǎn)者處理。然后,我們的消費(fèi)者將通過登錄到控制臺來捕獲并處理它。
@RestController @RequestMapping(value = <font>"/kafka"</font><font>) <b>public</b> <b>class</b> KafkaController { <b>private</b> <b>final</b> Producer producer; @Autowired KafkaController(Producer producer) { <b>this</b>.producer = producer; } @PostMapping(value = </font><font>"/publish"</font><font>) <b>public</b> <b>void</b> sendMessageToKafkaTopic(@RequestParam(</font><font>"message"</font><font>) String message) { <b>this</b>.producer.sendMessage(message); } } </font>
讓我們使用cURL將消息發(fā)送給Kafka:
curl -X POST -F 'message=test' http://localhost:9000/kafka/publish
以上是“在Spring Boot應(yīng)用程序中如何使用Apache Kafka”這篇文章的所有內(nèi)容,感謝各位的閱讀!希望分享的內(nèi)容對大家有幫助,更多相關(guān)知識,歡迎關(guān)注創(chuàng)新互聯(lián)行業(yè)資訊頻道!
當(dāng)前標(biāo)題:在SpringBoot應(yīng)用程序中如何使用ApacheKafka
文章地址:http://muchs.cn/article48/pioiep.html
成都網(wǎng)站建設(shè)公司_創(chuàng)新互聯(lián),為您提供ChatGPT、品牌網(wǎng)站制作、網(wǎng)站制作、關(guān)鍵詞優(yōu)化、小程序開發(fā)、全網(wǎng)營銷推廣
聲明:本網(wǎng)站發(fā)布的內(nèi)容(圖片、視頻和文字)以用戶投稿、用戶轉(zhuǎn)載內(nèi)容為主,如果涉及侵權(quán)請盡快告知,我們將會在第一時間刪除。文章觀點(diǎn)不代表本網(wǎng)站立場,如需處理請聯(lián)系客服。電話:028-86922220;郵箱:631063699@qq.com。內(nèi)容未經(jīng)允許不得轉(zhuǎn)載,或轉(zhuǎn)載時需注明來源: 創(chuàng)新互聯(lián)