SpringCloud 交互式查詢

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

@Autowired
private InteractiveQueryService interactiveQueryService;

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

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

如果有多個(gè)Kafka Streams應(yīng)用程序?qū)嵗谶\(yùn)行，則在以交互方式查詢它們之前，您需要確定哪個(gè)應(yīng)用程序?qū)嵗休d密鑰。InteractiveQueryService API提供了用于標(biāo)識(shí)主機(jī)信息的方法。

為了使它起作用，必須按如下所示配置屬性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
}