如何在Kafka Streams中实现多个分区的总计数

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 java.util.Arrays;
import java.util.Properties;

public class KafkaStreamsExample {
    public static void main(String[] args) {
        // 设置Kafka Streams应用程序的配置参数
        Properties props = new Properties();
        props.put("application.id", "kafka-streams-example");
        props.put("bootstrap.servers", "localhost:9092");
        props.put("key.serde", "org.apache.kafka.common.serialization.StringSerde");
        props.put("value.serde", "org.apache.kafka.common.serialization.StringSerde");

        // 创建一个流构建器
        StreamsBuilder builder = new StreamsBuilder();

        // 创建一个输入流
        KStream<String, String> inputStream = builder.stream("input-topic");

        // 按照分区键进行分组
        KTable<String, Long> countTable = inputStream
                .groupBy((key, value) -> value) // 这里以value作为分区键
                .count(Materialized.as("count-store")); // 计数并存储到一个KTable中

        // 将KTable转换为KStream
        KStream<String, Long> countStream = countTable.toStream();

        // 将计数结果写入输出主题
        countStream.to("output-topic", Produced.with(countStream.keySerde(), countStream.valueSerde()));

        // 创建Kafka Streams应用程序
        KafkaStreams streams = new KafkaStreams(builder.build(), props);

        // 启动应用程序
        streams.start();

        // 添加关闭钩子，以便在应用程序关闭时执行清理操作
        Runtime.getRuntime().addShutdownHook(new Thread(streams::close));
    }
}