交互式查询
Kafka Streams binder API 提供了一个名为 InteractiveQueryService 的类，用于交互式查询状态存储。你可以在应用程序中将其作为 Spring bean 进行访问。从应用程序中访问此 bean 的一种简便方法是使用 autowire 自动装配该 bean。
@Autowired
private InteractiveQueryService interactiveQueryService;
一旦你获得了这个 bean 的访问权限，你就可以查询你感兴趣的特定状态存储。请参见下文。
ReadOnlyKeyValueStore<Object, Object> keyValueStore =
interactiveQueryService.getQueryableStoreType("my-store", QueryableStoreTypes.keyValueStore());
在启动过程中，上述用于检索存储的方法调用可能会失败。例如，状态存储可能仍在初始化过程中。在这种情况下，重试此操作将非常有用。Kafka Streams binder 提供了一个简单的重试机制来适应这种情况。
以下是可用于控制此重试机制的两个属性。
-
spring.cloud.stream.kafka.streams.binder.stateStoreRetry.maxAttempts - 默认值为
1。 -
spring.cloud.stream.kafka.streams.binder.stateStoreRetry.backOffInterval - 默认值为
1000毫秒。
如果运行着多个 Kafka Streams 应用程序实例，那么在交互式查询之前，你需要确定哪个应用程序实例托管了你正在查询的特定键。InteractiveQueryService API 提供了用于识别主机信息的方法。
为了使此功能生效，您必须按如下方式配置 application.server 属性：
spring.cloud.stream.kafka.streams.binder.configuration.application.server: <server>:<port>
以下是一些代码片段：
org.apache.kafka.streams.state.HostInfo hostInfo = interactiveQueryService.getHostInfo("store-name",
key, keySerializer);
if (interactiveQueryService.getCurrentHostInfo().equals(hostInfo)) {
//query from the store that is locally available
}
else {
//query from the remote host
}
有关这些主机查找方法的更多信息，请参阅这些方法的Javadoc。同样，对于这些方法，在启动过程中，如果底层的KafkaStreams对象尚未就绪，它们可能会抛出异常。上述重试属性也适用于这些方法。
通过 InteractiveQueryService 可用的其他 API 方法
使用以下 API 方法检索与给定存储和密钥组合关联的 KeyQueryMetadata 对象。
public <K> KeyQueryMetadata getKeyQueryMetadata(String store, K key, Serializer<K> serializer)
使用以下 API 方法检索与给定存储和键组合关联的 KafkaStreams 对象。
public <K> KafkaStreams getKafkaStreams(String store, K key, Serializer<K> serializer)
自定义存储查询参数
有时，在通过 InteractiveQueryService 查询存储之前，需要微调存储查询参数。为此，从绑定器的 4.0.1 版本开始，你可以为 StoreQueryParametersCustomizer 提供一个 bean，这是一个函数式接口，其 customize 方法接受一个 StoreQueryParameter 作为参数。以下是其方法签名。
StoreQueryParameters<T> customize(StoreQueryParameters<T> storeQueryParameters);
采用这种方法，应用程序可以进一步自定义 StoreQueryParameters，例如启用陈旧存储。
当此 Bean 存在于应用程序中时，InteractiveQueryService 将在查询状态存储之前调用其 customize 方法。
请注意，应用程序中必须有一个唯一的 StoreQueryParametersCustomizer bean。