Spring Cloud 3.1关于如何在Spring boot应用程序中使用KTable的文档

Question

我很难找到任何可以使用Spring Cloud Streams的文档，这些文档将Kafka主题放入KTable中。已在此处查找文档，例如https://cloud.spring.io/spring-cloud-static/spring-cloud-stream-binder-kafka/3.0.0.RC1/reference/html/spring-cloud-stream-binder-kafka.html#_materializing_ktable_as_a_state_store在Spring boot中使用注释来实现这一点，没有什么是非常具体的。我希望我可以在我的应用程序中使用KStream创建一个简单的KTable。我有这样的属性：spring。云流动绑定。0中的进程。目的地：我的主题

然后在我的配置中，我希望我能做这样的事情

@Bean
public Consumer<KStream<String, String>> process() {
    return input -> input.toTable(Materialized.as("my-store"))
}

请告知我遗漏了什么？




Answer 1

如果您只想将来自Kafka主题的数据作为KTable使用，那么您可以按以下方式执行此操作。

@Bean
public Consumer<KTable<String, String>> process() {
    return input -> {
        
    };
}

如果要将表具体化为命名存储，则可以将其添加到配置中。

spring.cloud.stream.kafka.streams.bindings.process_in_0.consumer.materializedAs: my-store

您还可以执行问题中的操作，即将其作为KStream接收，然后转换为KTable。但是，如果这就是您需要做的全部，您可能更愿意首先将其作为KTable接收，就像我在这里建议的那样。