使用 Spring Boot 和 Kafka Streams 進行實時數據處理
Spring Boot 和 Apache Kafka Streams 是兩個強大的工具，它們使開發人員能夠創建可靠且可擴展的實時數據處理應用程序。在這篇文章中，我們將了解 Spring Boot 和 Kafka Streams 如何協同工作，如何利用流處理來發揮應用程序的優勢。還將探索交互式查詢，這是一個相對較新且有趣的功能，為實時數據分析提供了新的機會。
安裝Kafka
Kafka可以從官方網站https://kafka.apache.org/downloads下載。一旦 Kafka 啟動并運行，就創建一個主題。
創建Spring Boot項目
創建一個新的 Spring Boot 項目，并且引入“Spring Web”和“Spring for Apache Kafka”兩個依賴項。
@SpringBootApplication
public class KafkaStreamsDemoApplication {
public static void main(String[] args) {
SpringApplication.run(KafkaStreamsDemoApplication.class, args);
}
}配置Kafka
接下來，在應用程序的 application.properties 文件中配置 Kafka 創建的主題和代理地址。
spring.kafka.bootstrap-servers=localhost:9092
spring.kafka.consumer.group-id=my-group
spring.kafka.consumer.auto-offset-reset=earliest創建 Kafka 流處理器
下一步是構建一個 Kafka Streams 處理器，從“my-topic”讀取消息并處理，然后將結果輸出到另一個主題。使用 KStream API 來處理邏輯，如下：
@Bean
public Function<KStream<String, String>, KStream<String, String>> process() {
return input -> input
.mapValues(value -> value.toUpperCase())
.to("output-topic");
}交互式查詢
交互式查詢是 Kafka Streams 的創新新功能之一。借助此功能，可以立即查詢 Kafka Streams 應用程序的狀態存儲。讓我們看看如何使用交互式查詢檢索存儲在狀態存儲中的大寫消息的數量。
@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”的狀態存儲的句柄，然以查詢該存儲來獲取大寫消息的計數。
發送數據到Kafka
在實際應用程序中，數據將從多個源發送到 Kafka。在本示例中，我們將使用一個簡單的 Kafka 生產者來與“my-topic”進行通信。
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;
public void produceMessage(String message) {
kafkaTemplate.send("my-topic", message);
}使用處理后的數據
使用 Kafka 消費者最終從“output-topic”接收編輯后的數據，如下：
@KafkaListener(topics = "output-topic", groupId = "my-group")
public void consume(String message) {
System.out.println("Received: " + message);
}總結
在本文中，我們了解了如何使用 Spring Boot 和 Kafka Streams 創建用于實時數據處理的應用程序，并且引入了交互式查詢這一有趣的新功能。借助交互式查詢，可以通過處理實時數據以及實時查詢 Kafka Streams 應用程序的狀態來創建交互式動態應用程序。
