KStream筛选器是否会使用每条消息？

Question

我过去使用过Kafka，但从未使用过streams API。我的任务是构建一个可扩展的服务，该服务接受websocket连接，并根据用户id将出站消息从中心主题路由到正确的会话。

使用KStream看起来非常简单

builder.stream(inputTopic, Consumed.with(Serdes.String(), publicationSerde))
        .filter((name, publication) -> "George R. R. Martin".equals(publication.getName()))
        .to(outputTopic, Produced.with(Serdes.String(), publicationSerde));

但是筛选器命令是否消耗了主题中的每一条消息并在应用程序空间中执行筛选器？或者KStream

KStream上的措辞

如果过滤器的唯一目的是使用某个主题的每一条消息，并扔掉那些不相关的消息，我可以手动完成。

Answer 1

您是正确的-消息需要反序列化，然后根据谓词进行检查（在应用程序空间中）

扔掉那些不相关的东西，我可以手工完成

当然可以，但是Kafka Streams提供了定义会话窗口的有用方法。另外，您不需要定义消费者和生产者实例来转发新主题。