将 Spring 云流侦听器 (kafka) 从声明式迁移到函数式模型

Question

我正在尝试将 Spring Cloud Streams (kafka) 声明方式的实现迁移到推荐的功能模型

在this blog post 他们说：

...Spring Cloud Stream (SCSt) 中的函数式编程模型。它的 更少的代码，更少的配置。不过，最重要的是，您的代码是 完全解耦，独立于 SCSt 的内部

我目前的实现：

声明 MessageChanel

@Input(PRODUCT_INPUT_TOPIC)
MessageChannel productInputChannel();

使用现在已弃用的@StreamListener

@StreamListener(StreamConfig.PRODUCT_INPUT_TOPIC)
public void addProduct(@Payload Product product, @Header Long header1, @Header String header2)

Answer 1

在这里

@Bean
public Consumer<Product> addProduct() {
    return product -> {
        // your code
    };
}

我不确定PRODUCT_INPUT_TOPIC 的值是多少，但我们假设input。

所以 s-c-stream 会自动为您创建一个名为 addProduct-in-0 的绑定。 Here are the details。您可以按原样使用它，但如果您仍想使用自定义名称，您可以使用spring.cloud.stream.function.bindings.addProduct-in-0=input。 - 查看更多here。

如果您需要访问标头，您可以将消息作为输入参数传递

在这里

@Bean
public Consumer<Message<Product>> addProduct() {
    return message -> {
        Product product = message.getPayload();
        // your code
    };
}