Spring Boot 和 Apache Kafka Streams 是兩個強大的工具，它們使開發人員能夠創建可靠且可擴展的實時數據處理應用程序。在這篇文章中，我們將了解 Spring Boot 和 Kafka Streams 如何協同工作，如何利用流處理來發揮應用程序的優勢。還將探索交互式查詢，這是一個相對較新且有趣的功能，為實時數據分析提供了新的機會。zhs28資訊網——每日最新資訊28at.com

zhs28資訊網——每日最新資訊28at.com

安裝Kafka

Kafka可以從官方網站https://kafka.apache.org/downloads下載。一旦 Kafka 啟動并運行，就創建一個主題。zhs28資訊網——每日最新資訊28at.com

創建Spring Boot項目

創建一個新的 Spring Boot 項目，并且引入“Spring Web”和“Spring for Apache Kafka”兩個依賴項。zhs28資訊網——每日最新資訊28at.com

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

配置Kafka

接下來，在應用程序的 application.properties 文件中配置 Kafka 創建的主題和代理地址。zhs28資訊網——每日最新資訊28at.com

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

創建 Kafka 流處理器

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

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

交互式查詢

交互式查詢是 Kafka Streams 的創新新功能之一。借助此功能，可以立即查詢 Kafka Streams 應用程序的狀態存儲。讓我們看看如何使用交互式查詢檢索存儲在狀態存儲中的大寫消息的數量。zhs28資訊網——每日最新資訊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”的狀態存儲的句柄，然以查詢該存儲來獲取大寫消息的計數。zhs28資訊網——每日最新資訊28at.com

發送數據到Kafka

在實際應用程序中，數據將從多個源發送到 Kafka。在本示例中，我們將使用一個簡單的 Kafka 生產者來與“my-topic”進行通信。zhs28資訊網——每日最新資訊28at.com

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

使用處理后的數據

使用 Kafka 消費者最終從“output-topic”接收編輯后的數據，如下：zhs28資訊網——每日最新資訊28at.com

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

總結

在本文中，我們了解了如何使用 Spring Boot 和 Kafka Streams 創建用于實時數據處理的應用程序，并且引入了交互式查詢這一有趣的新功能。借助交互式查詢，可以通過處理實時數據以及實時查詢 Kafka Streams 應用程序的狀態來創建交互式動態應用程序。zhs28資訊網——每日最新資訊28at.com

本文鏈接：http://www.www897cc.com/showinfo-26-13310-0.html使用 Spring Boot 和 Kafka Streams 進行實時數據處理

聲明：本網頁內容旨在傳播知識，若有侵權等問題請及時與本網聯系，我們將在第一時間刪除處理。郵件：2376512515@qq.com

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

下一篇： C#.Net析構知識引申(CLR級的剖析）