Java的Kafka与分布式流处理

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KGroupedStream;
import org.apache.kafka.streams.kstream.Materialized;
import org.apache.kafka.streams.kstream.Produced;
 
import java.util.Arrays;
import java.util.Properties;
 
public class KafkaStreamsExample {
    public static void main(String[] args) {
        // 设置Kafka Streams配置
        Properties streamsConfiguration = new Properties();
        streamsConfiguration.put(StreamsConfig.APPLICATION_ID_CONFIG, "streams-wordcount");
        streamsConfiguration.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
        streamsConfiguration.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        streamsConfiguration.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());
 
        // 构建Kafka Streams顶ology
        StreamsBuilder builder = new StreamsBuilder();
        KGroupedStream<String, String> textLines = builder.stream("TextLinesTopic");
 
        textLines
            .flatMapValues(textLine -> Arrays.asList(textLine.toLowerCase().split("\\W+")))
            .groupBy((key, word) -> word, Grouped.with(Serdes.String(), Serdes.Long()))
            .count()
            .toStream()
            .to("WordsWithCountsTopic", Produced.with(Serdes.String(), Serdes.Long()));
 
        // 构建并启动Kafka Streams实例
        KafkaStreams streams = new KafkaStreams(builder.build(), streamsConfiguration);
        streams.start();
 
        // 处理ShutdownHook
        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
    }
}

这段代码展示了如何使用Kafka Streams库来进行简单的分布式流处理。它从一个名为"TextLinesTopic"的Kafka主题中读取文本行，将它们转换为小写单词，并统计每个单词出现的次数，然后将结果输出到另一个名为"WordsWithCountsTopic"的Kafka主题中。代码中包含了配置Kafka Streams实例和处理ShutdownHook的基本步骤。