我可以在 Kafka Stream Topology 中多次使用主题吗？答案

【问题标题】：Can I use topic more than once in Kafka Stream Topology?我可以在 Kafka Stream Topology 中多次使用主题吗？
【发布时间】：2020-08-06 21:54:18
【问题描述】：

让我们假设 groupby 函数在 kafka 流中不可用。我可以执行以下操作来获取字数并在其上构建 KTable 吗？请注意，我在拓扑中使用了两次“word-count-topic”。我有一个用例，我想迭代地构建一些东西，对于下一个流事件，我想查找以前的值并根据事件更新它。我想在我构建 Ktable 的同一主题中保留最新的价值。

KTable<String,Long> wordCountTable = builder.table("word-count-topic",Consumed.with(Serdes.String(), Serdes.Long()));

KStream<String,String> wordsStream = builder.stream("words-topic",Consumed.with(Serdes.String(), Serdes.String()));

KStream<String,String> msgStream = wordsStream
                                   .flatMapValues(textLine -> Arrays.asList(textLine.toLowerCase().split("\\W+")))
                                   .selectKey((k,v) -> v);

msgStream.leftJoin(kTable, (word,count) -> {
                                             if( count == null) return new WordCount(word, Long.valueOf(1));
                                             else return new WordCount(word, count + 1);
                                           })
            .mapValues((k,v)-> v.getCount())
            .to("word-count-topic", Produced.with(Serdes.String(), Serdes.Long()));

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

【问题讨论】：

这能回答你的问题吗？ Kafka Streams API: KStream to KTable
不，它没有。 “word-count-topic”最初是空的，在一段时间内，对于一个键，它有最新的更新。但是，更新过程是同一拓扑的一部分，它基于来自“words-topic”的流事件进行计算，并返回到“word-count-topic”并更新 Ktable 中的 prev 值。我可以这样做吗？

标签： apache-kafka kafka-consumer-api apache-kafka-streams ktable

【解决方案1】：

应该可以。为什么不直接运行代码？

【讨论】：

我运行了代码。它符合我的预期，但我只是想知道是否有任何警告？
好 :) -- 不应该有任何警告。