apache-kafka - Kafka Streams - 定义具有容错能力的自定义关系/Non_Key

问题描述

我正在尝试使用 kafka 实现事件溯源。

我对流处理器应用程序的设想是一个典型的 3 层 Spring 应用程序，其中：

“表示”层被（实现？）Kafka 流 API 所取代。
业务逻辑层由拓扑中的处理器 API 使用。
此外，该数据库是一个关系型 H2 内存数据库，可通过 Spring Data JPA 存储库访问。存储库还实现了必要的接口，以便将它们注册为 Kafka 状态存储以使用好处（恢复和容错）

但我想知道我应该如何实现自定义状态存储部分？

我一直在寻找并且：

有一些接口，例如StateStore& StoreBuilder。StoreBuilder有withLoggingEnabled()方法；但是，如果我启用它，实际更新和更改日志记录何时发生？通常，即使对于自定义示例，这些示例也是所有键值存储。如果我不想要键值怎么办？kafka 文档中交互式查询部分的示例并没有削减它。
我知道交互式查询。但它们似乎适用于查询而不是更新；顾名思义。

在键值存储中，发送到更改日志的记录很简单。但是如果我不使用键值；我何时以及如何通知 kafka 我的状态发生了变化？

标签： apache-kafkaapache-kafka-streamsspring-cloud-stream

您将需要为StateStore您要使用的实际存储引擎实现。这个界面没有任何关于商店的规定，你可以做任何你想做的事情。

您还需要实现一个StoreBuilder充当工厂的工厂来创建自定义商店的实例。

MyCustomStore implements StateStore {
    // define any interface you want to present to the user of the store
}

MyCustomStoreBuilder implements StoreBuilder<MyCustomStore> {
    MyCustomStore builder() {
        // create new instance of MyCustomStore and return it
    }

    // all other methods (except `name()`) are optional
    // eg, you can do a dummy implementation that only returns `this`
}

比较：https ://docs.confluent.io/current/streams/developer-guide/processor-api.html#implementing-custom-state-stores

但是如果我不使用键值；我何时以及如何通知 kafka 我的状态发生了变化？

如果要实现withLoggingEnabled()（类似于缓存），则需要将此日志记录（或缓存）实现为商店的一部分。因为，Kafka Streams 不知道您的商店是如何工作的，所以它无法为此提供实现。因此，这是您的设计决定，您的商店是否支持登录更改日志主题。如果您想支持日志记录，您需要提出一种设计，将存储更新映射到键值对（您也可以每次更新写入多个），您可以将其写入更改日志主题并允许您重新创建状态在从更改日志主题中读取这些记录时。

不仅可以通过更改日志来获得容错存储。例如，您还可以插入远程存储，它在内部进行复制等，因此依赖存储的容错能力而不是使用更改日志记录。当然，与使用本地商店相比，使用远程商店意味着其他挑战。

对于 Kafka Streams 默认存储，日志记录和缓存被实现为实际存储的包装器，使其易于插入。但是您可以以最适合您的商店的任何方式实现此功能。您可能需要查看以下类的键值存储作为比较：

对于交互式查询，您可以实现相应QueryableStoreType的以集成您的自定义商店。参照。https://docs.confluent.io/current/streams/developer-guide/interactive-queries.html#querying-local-custom-state-stores没错，交互式查询是现有商店的只读界面，因为Processors应负责维护商店。但是，也没有什么能阻止您打开自定义存储以进行写入。但是，这将使您的应用程序本质上是不确定的，因为如果您倒回输入主题并重新处理它，它可能会计算出不同的结果，具体取决于执行的“外部存储写入”。您应该考虑通过输入主题对商店进行任何写入。但这是你的决定。如果您允许“外部写入”，您将需要确保它们也被记录，以防您想要实现日志记录。

apache-kafka - Kafka Streams - 定义具有容错能力的自定义关系/Non_Key_Value StateStore

问题描述

解决方案

推荐阅读