InteractiveQueryService 在spring中，云数据流不可用于 StreamListener . 正如您提到的，它应该在主拓扑之外使用。但是，在您描述的用例中，您仍然可以使用主流中的状态存储。例如，如果您有一个传入的 KStream 和一个 KTable 具体化为状态存储，然后您可以调用 process 上 KStream 这样就可以进入国家商店。下面是一个粗略的代码来实现这一点。您需要将其转换为适合您的特定用例，但这里有一个想法。

ReadOnlyKeyValueStore<Object, String> store;

 input.process(() -> new Processor<Object, Product>() {

                @Override
                public void init(ProcessorContext processorContext) {
                    store = (ReadOnlyKeyValueStore) processorContext.getStateStore("my-store");

                }

                @Override
                public void process(Object key, Object value) {
                    //find the key
                    store.get(key);
                }

                @Override
                public void close() {
                    if (state != null) {
                        state.close();
                    }
                }
            }, "my-store");