paulwong

EVEN DRIVEN - SPRING CLOUD STREAM 3.x - Functional Programming Model

SPRING CLOUD STREAM 3.x 版本时,之前的一些编程模式,如@Enablebindding,@StreamListenner等注释被废弃了,这是由于一些框架的代码必需由用户编写,如配置框架用的Input MessageChannel,Output  MessageChannel,连接MessageHandler与MessageChannel等,被视为不必要的动作。为了简化用户代码,于是推出Functional Programming Model。

引入了新名词:Supplier、Function与Consumer。实际上这几个类可视为Adapter,如果之前已经有存在的Service类,且方法名为各种各样,可以重新包装成Supplier、Function与Consumer,并在固定的方法名:apply/get/accept中调用Service的方法。

Supplier

当在配置文件中注入此类型的Bean,并在spring.cloud.stream.function.definition加入此Bean的名称,SPRING CLOUD STREAM就会帮你生成一个Output  MessageChannel,并连接上此Bean,后续只需要在BINDDING中加入对应的Destination Name,即可向BROKER发消息了。

Consumer

当在配置文件中注入此类型的Bean,并在spring.cloud.stream.function.definition加入此Bean的名称,SPRING CLOUD STREAM就会帮你生成一个Input  MessageChannel,并连接上此Bean,后续只需要在BINDDING中加入对应的Destination Name,即可收到BROKER推送关于此Destination的消息了。

Function

当在配置文件中注入此类型的Bean,并在spring.cloud.stream.function.definition加入此Bean的名称,SPRING CLOUD STREAM就会帮你生成一个Input和Output  MessageChannel,并连接上此Bean,后续只需要在BINDDING中分别对Input和Output MessageChannel加入对应的Destination Name1/Name2,即可收到BROKER推送关于此Destination的消息,也可以向BROKER发消息了。

与SPRING INTEGRATION的整合

如果要对消息进行复杂处理,如拆分消息、聚合消息、IF ELSE消息等,就要借助SPRING INTEGRATION了。

@Bean
    public IntegrationFlow upperCaseFlow(LoanService loanService) {
        return IntegrationFlows
                    //turn this IntegrationFlow as a gateway, here is a Function interface 
                    
//with loadCheckerFunction as bean name
                    .from(LoadCheckerFunction.class, gateway -> gateway.beanName("loadCheckerFunction"))
                    .handle(loanService, "check")
                    .logAndReply(LoggingHandler.Level.WARN);
    }

    public interface LoadCheckerFunction extends Function<Loan, Loan>{

    }

IntegrationFlows.from(Class<?> serviceInterface)是可以将本IntegrationFlow包装成serviceInterface的实现类,如果调用此接口,最终会返回IntegrationFlow最后一个步骤的实体,如果这个serviceInterface是Function的话,刚好和SPRING CLOUD STREAM对接上。

后续在spring.cloud.stream.function.definition加入此Bean的名称loadCheckerFunction,SPRING CLOUD STREAM就会帮你生成一个Input和Output  MessageChannel,并连接上此Bean,再在BINDDING中分别对Input和Output MessageChannel加入对应的Destination Name1/Name2,即可收到BROKER推送关于此Destination的消息,也可以向BROKER发消息。

application.yaml
# This setting can increase or decrease the rate of message production (1000 = 1s)
# spring.cloud.stream.poller.fixed-delay=1000

# This setting can control which function method in our code will be triggered if there are multiple
# spring.cloud.function.definition=supplyLoan

# Give the autogenerated binding a friendlier name

spring:
   application:
      name: loan-check-rabbit
   banner:
      location: classpath:/banner-rabbit.txt
   cloud:
      stream:
         function.definition: loadCheckerFunction
         #BindingProperties
         bindings:
            loadCheckerFunction-in-0:
               destination: queue.pretty.log.messages
               binder: local_rabbit
            loadCheckerFunction-out-0:
               destination: queue.pretty.approved.messages
               binder: local_rabbit
         #BinderProperties
         binders:
            local_rabbit:
               type: rabbit
               environment:
                  spring:
                     rabbitmq:
                        host: 10.80.27.69
                        port: 5672
                        username: guest
                        password: guest
                        virtual-host: my-virtual-host

Reference

https://spring.io/blog/2019/10/25/spring-cloud-stream-and-spring-integration

posted on 2021-11-10 15:10 paulwong 阅读(364) 评论(0)  编辑  收藏 所属分类: SPRING INTERGRATIONSPRING CLOUDEVEN DRIVEN ARCHITECT


只有注册用户登录后才能发表评论。


网站导航: