所以你要做两件事：
答。服务停止时必须存储以下参数：
如果您想以一种简单的方式完成它，那么您所要做的就是在与状态存储相关联的主题中编写一条消息（您正在使用 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));

然后在转换器或处理器中使用它来添加项目；您必须在转换器/处理器中声明：

// 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流你首先要让商店可用；最简单的方法是创建一个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);