Spring Kafka Streams - Create Stream vs Inject an existing Stream bean

Question

我正在使用Kafka Streams创建一个基于Spring的服务，并注意到当我使用现有的@Bean KStream与为该流创建流时，会有很大的耗时差异。

假设我有以下代码：

@Autowired
private StreamsBuilder eventsStreamsBuilder;

@Bean("eventsKStream")
public KStream<String, String> eventsKStream() {

   KStream<String, String> stream = eventsStreamsBuilder.stream(...);
   stream.to("NEW_TOPIC");
   return stream;
}

// 1. First approach, create a stream for the "NEW_TOPIC" and filter it
@Bean("eventsFilterKStream")
public KStream<String, String> eventsFilterKStream() {
   return eventsStreamsBuilder.stream("NEW_TOPIC",....)
   .filter()
}

// 2. Second approach, inject the existing Stream Bean and filter it
@Bean("eventsFilterKStream")
public KStream<String, String> eventsFilterKStream(@Qualifier("eventsKStream") KStream<String, String> eventsKStream) {
   return eventsKStream.filter(...);
}

我有一个消费者，正在使用控制台消费者...并注意到，对于第一种方法，数据需要几秒钟才能到达，但在第二种方法中，我会立即获得过滤后的数据！

你能给我解释一下有什么区别吗？为什么第一种方法需要几秒钟时间才能将消息到达消费者那里？最好的方法是什么？

谢谢!

Answer 1

方法#2更快，因为它向现有的KStream添加了一个过滤器处理器节点，而不是在方法#1中从源主题创建一个全新的KStream读取。

这取决于你的拓扑结构，但我会从#2开始，除非你有很强的理由不这么做。