apache-kafka-streams - 如何将 KStream 注入 Spring bean？

Question

我正在使用 Spring Cloud Stream 和 Kafka 流开发 Spring Boot 应用程序。我们有几个函数式的消息生产者、消费者和函数，SpCS 的魔力将KStream实例连接到我们的函数方法中。

但是，现在我需要编写一个类，它可以过滤其中一个流（基于某些输入）并收集事件以返回给调用者。基本上，一个端点将获取一个键作为参数，并需要从具有该键的流中返回事件。

问题是，我不知道如何将KStream实例放入该组件中。我尝试了（对我而言）显而易见的事情，只是KStream直接注入，但 Spring 告诉我上下文中没有这样的 bean。我对此并不完全感到惊讶，但问题是KStream当我不在功能/消费 bean 中时，如何获得对我想要过滤/转换/任何内容的引用？

或者，也许我在吠叫完全错误的树，并且有一种不同的、流式思维方式的解决方案来满足需求，这与我试图实现的伪存储库不同。可以对所有事件运行参数化“查询”（又名过滤器）并收集它们（不仅仅是最新的）的东西？

标签： apache-kafka-streamsspring-kafkaspring-cloud-stream