Spring Boot 和 Apache Kafka Streams 是兩個(gè)強(qiáng)大的工具，它們使開發(fā)人員能夠創(chuàng)建可靠且可擴(kuò)展的實(shí)時(shí)數(shù)據(jù)處理應(yīng)用程序。在這篇文章中，我們將了解 Spring Boot 和 Kafka Streams 如何協(xié)同工作，如何利用流處理來發(fā)揮應(yīng)用程序的優(yōu)勢。還將探索交互式查詢，這是一個(gè)相對較新且有趣的功能，為實(shí)時(shí)數(shù)據(jù)分析提供了新的機(jī)會(huì)。cpI28資訊網(wǎng)——每日最新資訊28at.com

cpI28資訊網(wǎng)——每日最新資訊28at.com

安裝Kafka

Kafka可以從官方網(wǎng)站https://kafka.apache.org/downloads下載。一旦 Kafka 啟動(dòng)并運(yùn)行，就創(chuàng)建一個(gè)主題。cpI28資訊網(wǎng)——每日最新資訊28at.com

創(chuàng)建Spring Boot項(xiàng)目

創(chuàng)建一個(gè)新的 Spring Boot 項(xiàng)目，并且引入“Spring Web”和“Spring for Apache Kafka”兩個(gè)依賴項(xiàng)。cpI28資訊網(wǎng)——每日最新資訊28at.com

@SpringBootApplicationpublic class KafkaStreamsDemoApplication {    public static void main(String[] args) {        SpringApplication.run(KafkaStreamsDemoApplication.class, args);    }}

配置Kafka

接下來，在應(yīng)用程序的 application.properties 文件中配置 Kafka 創(chuàng)建的主題和代理地址。cpI28資訊網(wǎng)——每日最新資訊28at.com

spring.kafka.bootstrap-servers=localhost:9092spring.kafka.consumer.group-id=my-groupspring.kafka.consumer.auto-offset-reset=earliest

創(chuàng)建 Kafka 流處理器

下一步是構(gòu)建一個(gè) Kafka Streams 處理器，從“my-topic”讀取消息并處理，然后將結(jié)果輸出到另一個(gè)主題。使用 KStream API 來處理邏輯，如下：cpI28資訊網(wǎng)——每日最新資訊28at.com

@Beanpublic Function<KStream<String, String>, KStream<String, String>> process() {    return input -> input            .mapValues(value -> value.toUpperCase())            .to("output-topic");}

交互式查詢

交互式查詢是 Kafka Streams 的創(chuàng)新新功能之一。借助此功能，可以立即查詢 Kafka Streams 應(yīng)用程序的狀態(tài)存儲(chǔ)。讓我們看看如何使用交互式查詢檢索存儲(chǔ)在狀態(tài)存儲(chǔ)中的大寫消息的數(shù)量。cpI28資訊網(wǎng)——每日最新資訊28at.com

@Autowiredprivate InteractiveQueryService interactiveQueryService;@GetMapping("/messageCount")public long getMessageCount() {  ReadOnlyKeyValueStore<String, Long> store = interactiveQueryService.getQueryableStore("message-count-store", QueryableStoreTypes.keyValueStore());     return store.get("uppercase-message-count");}

在此代碼中，我們使用 InteractiveQueryService 來獲取“message-count-store”的狀態(tài)存儲(chǔ)的句柄，然以查詢該存儲(chǔ)來獲取大寫消息的計(jì)數(shù)。cpI28資訊網(wǎng)——每日最新資訊28at.com

發(fā)送數(shù)據(jù)到Kafka

在實(shí)際應(yīng)用程序中，數(shù)據(jù)將從多個(gè)源發(fā)送到 Kafka。在本示例中，我們將使用一個(gè)簡單的 Kafka 生產(chǎn)者來與“my-topic”進(jìn)行通信。cpI28資訊網(wǎng)——每日最新資訊28at.com

@Autowiredprivate KafkaTemplate<String, String> kafkaTemplate;public void produceMessage(String message) {    kafkaTemplate.send("my-topic", message);}

使用處理后的數(shù)據(jù)

使用 Kafka 消費(fèi)者最終從“output-topic”接收編輯后的數(shù)據(jù)，如下：cpI28資訊網(wǎng)——每日最新資訊28at.com

@KafkaListener(topics = "output-topic", groupId = "my-group")public void consume(String message) {    System.out.println("Received: " + message);}

總結(jié)

在本文中，我們了解了如何使用 Spring Boot 和 Kafka Streams 創(chuàng)建用于實(shí)時(shí)數(shù)據(jù)處理的應(yīng)用程序，并且引入了交互式查詢這一有趣的新功能。借助交互式查詢，可以通過處理實(shí)時(shí)數(shù)據(jù)以及實(shí)時(shí)查詢 Kafka Streams 應(yīng)用程序的狀態(tài)來創(chuàng)建交互式動(dòng)態(tài)應(yīng)用程序。cpI28資訊網(wǎng)——每日最新資訊28at.com

本文鏈接：http://www.tebozhan.com/showinfo-26-13310-0.html使用 Spring Boot 和 Kafka Streams 進(jìn)行實(shí)時(shí)數(shù)據(jù)處理

聲明：本網(wǎng)頁內(nèi)容旨在傳播知識(shí)，若有侵權(quán)等問題請及時(shí)與本網(wǎng)聯(lián)系，我們將在第一時(shí)間刪除處理。郵件：2376512515@qq.com

上一篇： 推薦 13 個(gè) IntelliJ IDEA 高手代碼編輯技巧！

下一篇： C#.Net析構(gòu)知識(shí)引申(CLR級的剖析）