SpringCloud 交互式查詢

作為公共Kafka Streams綁定程序API的一部分，我們公開了一個名為InteractiveQueryService的類。您可以在應(yīng)用程序中以Spring bean的身份進行訪問。從您的應(yīng)用程序訪問此bean的一種簡單方法是“自動裝配” bean。

@Autowired
private InteractiveQueryService interactiveQueryService;

一旦獲得對此bean的訪問權(quán)限，就可以查詢您感興趣的特定狀態(tài)存儲。見下文。

ReadOnlyKeyValueStore<Object, Object> keyValueStore =
						interactiveQueryService.getQueryableStoreType("my-store", QueryableStoreTypes.keyValueStore());

如果有多個Kafka Streams應(yīng)用程序?qū)嵗谶\行，則在以交互方式查詢它們之前，您需要確定哪個應(yīng)用程序?qū)嵗休d密鑰。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
}