写入 Kafka Streams 上的 GlobalStateStore答案

【问题标题】：Write to GlobalStateStore on Kafka Streams写入 Kafka Streams 上的 GlobalStateStore
【发布时间】：2019-05-21 13:58:44
【问题描述】：

我正在尝试在 Kafka DSL 上使用 addGlobalStore，在该 DSL 中需要存储一些我需要对所有线程/实例进行全局访问的值。

我的问题是我需要定期更新拓扑中的这些值，并使所有正在运行的线程都知道新值。

我通过 builder.addGlobalStore 初始化全局存储，并使用处理器的 init() 函数用作此函数的最后一个参数，但我找不到更新全局存储内值的方法。

我的 Topology 的下一步是一个 Transformer，我可以通过全局 Store 上的 ```init()` 获取一个钩子并读取存储的值，但不幸的是我无法全局更新它们。我的意思是我可以更新正在运行的线程的本地副本，但其他线程/实例看不到更改。

我在某处读到，这不能在 Transformer 上完成，但即使我使用处理器，问题仍然存在

那么，有没有办法在 Kafka DSL 拓扑上更新 globalStateStore，如果是这样，这怎么可能？或者为了使用全局存储，我需要使用低级处理器 API 吗？

【问题讨论】：

【解决方案1】：

我通过 builder.addGlobalStore 初始化全局存储，并使用处理器的 init() 函数用作此函数的最后一个参数，但我找不到更新全局存储内值的方法。

您不能直接更新全局存储。相反，您必须更新（= 向其写入消息）该全局存储的基础主题。

【讨论】：

【解决方案2】：

如果它符合您的需求，您可能可以使用 GlobalKTable 而不是 GlobalStore

【讨论】：