Kafka Streams - 使用处理器 Api 实现加入

Question

我知道可以使用 dsl api 执行连接。由于各种原因，我们需要使用处理器 api。

如何使用处理器 api 实现加入流。我有一些想法，但不认为它们是正确的。

1. 具有多个源主题的一个处理器。流程接口的基础对象，然后在流程方法中转换为正确的类型。

2. 两个处理器，每个处理器都有自己的源主题。每个处理器都获得对其他处理器状态存储的只读访问（如果可能）。

任何想法 - 我确实在 KStreamImpl 中找到了 join 实现，但在遵循时遇到了问题。也许是关于 dsl 如何做到的解释？

Answer 1

您建议的两种实现都是可能的。Kafka Stream 本身使用 5 个处理器来实现流-流连接：

source1 ---> "state maintainer 1" --> "joiner 1" ----+
                      |                   |          |
                   updates          "join lookups"   |
                      |                   |          +-----+
                      |            +------+                |
                      v            |                       v
                  "state 1" <------|------+             "merger" -->
                                   |      |                ^
                  "state 2" <------+      |                |
                      ^                   |          +-----+
                      |                   |          |
                   updates          "join lookups"   |
                      |                   |          |
source2 ---> "state maintainer 2" --> "joiner 2" ----+

左右管道对称。两者都有一个“状态维护者”和“加入者” Processor。“状态维护者”对状态有写访问权限。“Joiner”作为对其他状态的读访问。最后一步，将两个连接结果流合并在一起。