将数据添加到状态存储以进行有状态处理和容错

Question

我有一个执行一些有状态处理的微服务。应用程序从输入主题构建 KStream，进行一些有状态处理，然后将数据写入输出主题。

我将运行 3 个此应用程序放在同一组中。当微服务出现故障时，我需要存储 3 个参数，接管的微服务可以查询共享状态存储并从崩溃服务停止的地方继续。

我正在考虑将这 3 个参数推送到 statestore 中，并在其他微服务接管时查询数据。从我的研究中，我看到了很多人们使用状态存储执行事件计数的例子，但这并不是我想要的，有没有人知道一个例子或者解决这个问题的正确方法是什么？

Answer 1

所以你想做两件事：

一个。服务中断必须存储参数：
如果你想以一种直接的方式来做，那么你所要做的就是在与状态存储相关的主题中写一条消息（你正在用 KTable 阅读的那个）。使用 Kafka Producer API 或 KStream（可能是 kTable.toStream()）就可以了。

否则您可以手动创建状态存储：

// take these serde as just an example
Serde<String> keySerde = Serdes.String();
Serde<String> valueSerde = Serdes.String();
KeyValueBytesStoreSupplier storeSupplier = inMemoryKeyValueStore(stateStoreName);
streamsBuilder.addStateStore(Stores.keyValueStoreBuilder(storeSupplier, keySerde, valueSerde));

然后在变压器或处理器中使用它来添加项目；您必须在 transformer/processor:

中声明

// depending on the serde above you might have something else then String
private KeyValueStore<String, String> stateStore;

并初始化stateStore变量：

@Override
public void init(ProcessorContext context) {
  stateStore = (KeyValueStore<String, String>) context.getStateStore(stateStoreName);
}

稍后使用 stateStore 变量：

@Override
public KeyValue<String, String> transform(String key, String value) {
  // using stateStore among other actions you might take here
  stateStore.put(key, processedValue);
}

b。读取接管服务中的参数：
您可以使用 Kafka 消费者来完成此操作，但使用 Kafka Streams，您首先必须使商店可用；最简单的方法是创建一个 KTable；那么你必须获得使用 KTable 自动创建的可查询商店名称；然后你必须实际访问商店；然后你从商店中提取一个记录值（即通过其键的参数值）。

// this example is a modified copy of KTable javadocs example
final StreamsBuilder streamsBuilder = new StreamsBuilder();

// Creating a KTable over the topic containing your parameters a store shall automatically be created.
//
// The serde for your MyParametersClassType could be 
// new org.springframework.kafka.support.serializer.JsonSerde(MyParametersClassType.class) 
// though further configurations might be necessary here - e.g. setting the trusted packages for the ObjectMapper behind JsonSerde.
//
// If the parameter-value class is a String then you could use Serdes.String() instead of a MyParametersClassType serde.
final KTable paramsTable = streamsBuilder.table("parametersTopicName", Consumed.with(Serdes.String(), <<your InstanceOfMyParametersClassType serde>>));

...
// see the example from KafkaStreams javadocs for more KafkaStreams related details
final KafkaStreams streams = ...;
streams.start()
...

// get the queryable store name that is automatically created with the KTable
final String queryableStoreName = paramsTable.queryableStoreName();
// get access to the store
ReadOnlyKeyValueStore view = streams.store(queryableStoreName, QueryableStoreTypes.timestampedKeyValueStore());
// extract a record value from the store
InstanceOfMyParametersClassType parameter = view.get(key);

将数据添加到状态存储以进行有状态处理和容错

Adding data to state store for stateful processing and fault tolerance

stateful

apache-kafka

rocksdb

apache-kafka-streams