如何在Spring-Cloud-Stream中使用kafka流程拓扑内的交互查询？

Question

是否可以在Spring Cloud中使用@EnableBinding注释的类流或在方法中使用@StreamListener使用交互式查询（InteractiveQueryService）？我尝试在提供的KStreamMusicSampleApplication类和process方法中实例化ReadOnlyKeyValueStore，但它始终为空。

我的@StreamListener方法正在监听一组KTables和KStreams，在进程拓扑（例如过滤）期间，我必须检查来自KStream的密钥是否已经存在于特定的KTable中。

我试图弄清楚如何扫描传入的KTable来检查密钥是否已经存在，但没有运气。然后我遇到了InteractiveQueryService,它的get（）方法可以用于检查KTable中的状态存储materializedAs中是否存在键。问题是我无法使用进程拓扑(@enablebinding或@streamlistener)访问它。只能从这些注释（例如RestController）外部访问它。

有没有一种方法可以扫描传入的KTable来检查键或值的存在？如果不能，那么我们可以访问流程拓扑中的InteractiveQueryService吗？

Answer 1

Spring Cloud Stream中的InteractiveQueryService无法在StreamListener的实际拓扑中使用。正如您提到的，它应该在主拓扑之外使用。但是，使用您描述的用例，您仍然可以使用主流中的状态存储。例如，如果您有一个传入的KStream和一个被物化为状态存储的KTable，那么您可以调用KStream上的Process并以这种方式访问状态存储。下面是一个实现这一点的粗略代码。您需要将其转换为适合您的特定用例，但这里是想法。

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");