代码之家 › 专栏 › 技术社区 › Cemo

如何获得有关GlobalKTable状态存储更新的通知?

apache-kafka-streams

Cemo · 技术社区 · 8 年前

我使用StreamsBuilder的simply API构建GlobalKTable,如下所示:

Materialized<Long, Category, KeyValueStore<Bytes, byte[]>> materialized =
    Materialized.<Long, Category, KeyValueStore<Bytes, byte[]>>as(this.categoryStoreName)
        .withCachingDisabled()
        .withKeySerde(Serdes.Long())
        .withValueSerde(CATEGORY_JSON_SERDE);

return streamsBuilder.globalTable(categoryTopic, materialized);

我希望收到变更通知。它很少更新,如果更新,我想触发缓存失效。卡夫卡的做法是什么?

1 回复 | 直到 7 年前

Matthias J. Sax 8 年前

GlobalKTable不支持此操作。但是,您可以使用“全局存储”并实现您的自定义 Processor 每次更新都会调用。

内部,a GlobalKTable 使用“全局存储”,并提供 加工机 为您实施。

您可以通过以下方式添加全球门店: StreamsBuilder#addGlobalStore() 。

推荐文章

Stella · Kafka流处理期间处理数据时出现异常

8 年前

Stella · kafka流中的处理器节点

8 年前

mukesh210 · 卡夫卡流:Lib Rocks数据库上的UnsatifiedLinkError

8 年前