Spring Boot与Apache 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.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
 
@Configuration
public class KafkaStreamsConfig {
 
    @Bean
    public KafkaStreams kafkaStreams(StreamsBuilder streamsBuilder) {
        StreamsBuilder builder = new StreamsBuilder();
        KStream<String, String> stream = builder.stream("input-topic");
 
        stream.foreach((key, value) -> System.out.println(value));
 
        KafkaStreams kafkaStreams = new KafkaStreams(builder.build(), streamsConfig());
        kafkaStreams.start();
 
        return kafkaStreams;
    }
 
    private java.util.Map<String, Object> streamsConfig() {
        java.util.Map<String, Object> props = new HashMap<>();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-application");
        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        props.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        props.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        return props;
    }
}

这个代码示例展示了如何在Spring Boot应用程序中配置和启动Apache Kafka Streams。它定义了一个名为KafkaStreamsConfig的配置类，并在其中创建了一个KafkaStreams bean。这个bean包含了从配置的输入主题读取消息的逻辑，并且会将接收到的消息内容打印到控制台。这是一个简单的示例，展示了如何将Spring Boot和Kafka Streams集成在一起。