Spring Boot與Kafka Streams集成

Spring Boot與Kafka Streams集成可以讓你在Spring Boot應用程序中輕松地使用Kafka Streams。以下是一些關鍵步驟和代碼示例，幫助你完成這個集成。

1. 添加依賴

首先，在你的pom.xml文件中添加Spring Boot和Kafka Streams的依賴：

<dependencies>
    <!-- Spring Boot Starter Web -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-web</artifactId>
    </dependency>

    <!-- Kafka Streams -->
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-streams</artifactId>
    </dependency>

    <!-- Kafka Client -->
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
    </dependency>
</dependencies>

2. 配置Kafka

在你的application.yml或application.properties文件中配置Kafka連接信息：

spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: my-group
    producer:
      key-serializer: org.apache.kafka.common.serialization.StringSerializer
      value-serializer: org.apache.kafka.common.serialization.StringSerializer

3. 創建Kafka Streams配置類

創建一個配置類來設置Kafka Streams的配置：

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.KTable;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.Produced;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

@Configuration
public class KafkaStreamsConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Bean
    public KafkaStreams kafkaStreams() {
        StreamsBuilder builder = new StreamsBuilder();

        // 從輸入主題讀取數據
        KStream<String, String> inputStream = builder.stream("input-topic");

        // 處理數據（例如，過濾和轉換）
        KTable<String, String> processedTable = inputStream
            .filter((key, value) -> value.contains("example"))
            .mapValues(value -> value + "-processed");

        // 將處理后的數據寫入輸出主題
        processedTable.toStream()
            .to("output-topic", Produced.with(Serdes.String(), Serdes.String()));

        KafkaStreams streams = new KafkaStreams(builder.build(), getStreamsConfig());
        streams.start();

        return streams;
    }

    private KafkaStreams.Config getStreamsConfig() {
        return KafkaStreams.defaultConfig()
            .withBootstrapServers(bootstrapServers)
            .withApplicationId("my-kafka-streams-app");
    }
}

4. 啟動Spring Boot應用程序

在你的主應用程序類中啟動Kafka Streams：

import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

@SpringBootApplication
public class KafkaStreamsApplication {

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

5. 測試集成

你可以使用Kafka自帶的工具（如kafka-console-producer和kafka-console-consumer）來測試你的集成。

生產者

kafka-console-producer --broker-list localhost:9092 --topic input-topic

消費者

kafka-console-consumer --bootstrap-server localhost:9092 --topic output-topic --from-beginning

通過以上步驟，你已經成功地將Spring Boot與Kafka Streams集成在一起。你可以根據需要擴展和修改這個示例，以滿足你的具體需求。