@EnableBinding@从3.1开始被弃用，转而支持函数式编程模型

Question

我看到下面的注释是为春云流贬值

< code > @ Input < code > @ Output < code > @ enable binding < code > @ StreamListener

请提供示例和文档链接，说明如何以功能方式进行操作。

Answer 1

以下是一些更有帮助的信息：

发送消息

使用org.springframework.cloud.stream.function。用于发送消息的StreamBridge。

以前

myDataSource.output().send(message);

之后

streamBridge.send("myData-out-0", message);

替换ServiceActivator

以前

@ServiceActivator(inputChannel = MyProcessor.INPUT, outputChannel = MyProcessor.OUTPUT)
public Message<MySuperOutputMessage> transform(Message<MySuperInputMessage> message) { ... }

之后

@Bean
Function<Message<MySuperInputMessage>, Message<MySuperOutputMessage>> myCoolFunction() {
    return message -> {...}; 
}

不要忘记在属性spring.cloud.function.definition中注册“myCoolFunction”。

Answer 2

这个github存储库包含了很多例子。
https://github.com/spring-cloud/stream-applications

官方留档详细解释了如何从命令式到功能式在Spring云流应用程序与kafka流，但没有它是一样的。

https://docs.spring.io/spring-cloud-stream/docs/3.1.0/reference/html/spring-cloud-stream.html#spring_cloud_function

也请查看此帖子..
https://spring.io/blog/2019/10/14/spring-cloud-stream-demystified-and-simplified

https://docs.spring.io/spring-cloud-stream-binder-kafka/docs/3.1.0/reference/html/spring-cloud-stream-binder-kafka.html#_programming_model

有一个命令式代码的示例(https://docs.spring.io/spring-cloud-stream-binder-kafka/docs/3.1.0/reference/html/spring-cloud-stream-binder-kafka.html#_imperative_programming_model)以及如何以功能风格开发。

Answer 3

Spring现在使用检测到的消费者/功能/供应商bean来为您定义流，而不是使用基于注释的配置。旧版本带有注释的代码如下所示：

interface InputChannels {
    
        @Input("input")
        SubscribableChannel input();
    }

    @EnableBinding(InputChannels.class)
    public class PubSubDemo {
    @StreamListener("input")
    public void listen() {
            
        if (LOG.isInfoEnabled()) {
            LOG.info(context.toString());
        }
    }

新版本代码将类似于:

public class PubSubDemo {
   @Bean
   Consumer<String> input() {
    return str -> {
        
        if (LOG.isInfoEnabled()) {
            LOG.info(context.toString());
             }
        };
   }
}

检查消费者 bean 替换了@StreamListener和@Input。

关于配置，如果之前为了配置你有一个application.yml看起来像这样:

spring:
  cloud:
    stream:
      bindings:
        input:
          destination: destination
          group: group
          consumer:
            concurrency: 10
            max-attempts: 3

现在新的配置将是这样的

spring:
  cloud:
    stream:
      bindings:
        input-in-0:
          destination: destination
          group: group
          consumer:
            concurrency: 10
            max-attempts: 3

可选：Spring 云流库自行创建 bean。如果未创建 Bean。使用以下属性。

spring:
   cloud:
     stream:
        function:
           definition: input

输入和输出对应于绑定的类型（例如输入或输出）。索引是输入或输出绑定的索引。对于典型的单输入/输出功能，它始终为0。

现在让我们考虑输出通道：

    public interface OutputChannels {
        @Output
        MessageChannel output();
    }

    @Service
    @EnableBinding(OutputChannels.class)
    class PubSubSendQueue {
    
    OutputChannels  outputChannel;

    public void publish() {
        outputChannel.output().send("Hello");
    }
}

现在使用函数代码将作为：

@Service
class PubSubSendQueue {
    @Bean
    public Supplier<String> output(){
        return Supplier { "Adam" }
    }
}

在application.properties文件中添加以下内容

spring.cloud.stream.bindings.output-out-0.destination=destination