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集成在一起。你可以根据需要扩展和修改这个示例，以满足你的具体需求。