干货｜Spring Cloud Stream 体系及原理介绍

朱雀 2022-01-16 09:27 255阅读 0赞

[2019独角兽企业重金招聘Python工程师标准>>> ][2019_Python_] ![hot3.png][]

[Spring Cloud Stream][] 在 Spring Cloud 体系内用于构建高度可扩展的基于事件驱动的微服务，其目的是为了简化消息在 Spring Cloud 应用程序中的开发。Spring Cloud Stream (后面以 **SCS** 代替 Spring Cloud Stream) 本身内容很多，而且它还有很多外部的依赖，想要熟悉 SCS，必须要先了解 Spring Messaging 和 Spring Integration 这两个项目，接下来文章将从以下几点跟大家进行介绍：

*  什么是 Spring Messaging；
 *  什么是 Spring Integration；
 *  什么是 SCS及其功能；

### Spring Messaging ###

Spring Messaging 是 Spring Framework 中的一个模块，其作用就是统一消息的编程模型。

*  比如消息 `Messaging` 对应的模型就包括一个消息体 Payload 和消息头 Header:

![f17d6b9b23908b3e24798eb0cd4c3b61479.jpg][]

package org.springframework.messaging;
    public interface Message<T> {
        T getPayload();
        MessageHeaders getHeaders();
    }

*  消息通道 `MessageChannel` 用于接收消息，调用 `send` 方法可以将消息发送至该消息通道中 :

![4beb01d755ca4a3b0b3cf224259f3fda00a.jpg][]

@FunctionalInterface
    public interface MessageChannel {
        long INDEFINITE_TIMEOUT = -1;
        default boolean send(Message<?> message) {
            return send(message, INDEFINITE_TIMEOUT);
        }
        boolean send(Message<?> message, long timeout);
    }

消息通道里的消息如何被消费呢？

*  由消息通道的子接口可订阅的消息通道 `SubscribableChannel` 实现，被 `MessageHandler` 消息处理器所订阅:

public interface SubscribableChannel extends MessageChannel {
        boolean subscribe(MessageHandler handler);
        boolean unsubscribe(MessageHandler handler);
    }

*  由`MessageHandler` 真正地消费/处理消息:

@FunctionalInterface
    public interface MessageHandler {
        void handleMessage(Message<?> message) throws MessagingException;
    }

Spring Messaging 内部在消息模型的基础上衍生出了其它的一些功能，如：

1.  消息接收参数及返回值处理：消息接收参数处理器 `HandlerMethodArgumentResolver` 配合 `@Header`, `@Payload` 等注解使用；消息接收后的返回值处理器 `HandlerMethodReturnValueHandler` 配合 `@SendTo` 注解使用；
2.  消息体内容转换器 `MessageConverter`；
3.  统一抽象的消息发送模板 `AbstractMessageSendingTemplate`；
4.  消息通道拦截器 `ChannelInterceptor`；

### Spring Integration ###

[Spring Integration][] 提供了 Spring 编程模型的扩展用来支持企业集成模式([Enterprise Integration Patterns][])，是对 Spring Messaging 的扩展。它提出了不少新的概念，包括消息的路由 `MessageRoute`、消息的分发 `MessageDispatcher`、消息的过滤 `Filter`、消息的转换 `Transformer`、消息的聚合 `Aggregator`、消息的分割 `Splitter` 等等。同时还提供了 `MessageChannel` 和`MessageHandler` 的实现，分别包括 `DirectChannel`、`ExecutorChannel`、`PublishSubscribeChannel` 和`MessageFilter`、`ServiceActivatingHandler`、`MethodInvokingSplitter` 等内容。

首先为大家介绍几种消息的处理方式：

*  消息的分割：

![a89cb59cd3bf1e211977248597510544cfe.jpg][]

*  消息的聚合：

![3539199a64eed63ba5a01bb8d3e12ef674b.jpg][]

*  消息的过滤：

![2ae2420ec57395d2d617b8e3526f5dd2f76.jpg][]

*  消息的分发：

![58f4e010cf3b8eaef2853acd584563104d6.jpg][]

接下来，我们以一个最简单的例子来尝试一下 Spring Integration:

SubscribableChannel messageChannel = new DirectChannel(); // 1
    
    messageChannel.subscribe(msg -> { // 2
      System.out.println("receive: " + msg.getPayload());
    });
    
    messageChannel.send(MessageBuilder.withPayload("msg from alibaba").build()); // 3

1.  构造一个可订阅的消息通道 `messageChannel`；
2.  使用 `MessageHandler` 去消费这个消息通道里的消息；
3.  发送一条消息到这个消息通道，消息最终被消息通道里的 `MessageHandler` 所消费，最后控制台打印出: `receive: msg from alibaba`；

`DirectChannel` 内部有个 `UnicastingDispatcher` 类型的消息分发器，会分发到对应的消息通道 `MessageChannel` 中，从名字也可以看出来，`UnicastingDispatcher` 是个单播的分发器，只能选择一个消息通道。那么如何选择呢? 内部提供了 `LoadBalancingStrategy` 负载均衡策略，默认只有轮询的实现，可以进行扩展。

我们对上段代码做一点修改，使用多个 `MessageHandler` 去处理消息：

SubscribableChannel messageChannel = new DirectChannel();
    
    messageChannel.subscribe(msg -> {
      System.out.println("receive1: " + msg.getPayload());
    });
    
    messageChannel.subscribe(msg -> {
      System.out.println("receive2: " + msg.getPayload());
    });
    
    messageChannel.send(MessageBuilder.withPayload("msg from alibaba").build());
    messageChannel.send(MessageBuilder.withPayload("msg from alibaba").build());

由于 `DirectChannel` 内部的消息分发器是 `UnicastingDispatcher` 单播的方式，并且采用轮询的负载均衡策略，所以这里两次的消费分别对应这两个 `MessageHandler`。控制台打印出：

receive1: msg from alibaba
    receive2: msg from alibaba

既然存在单播的消息分发器 `UnicastingDispatcher`，必然也会存在广播的消息分发器，那就是 `BroadcastingDispatcher`，它被 `PublishSubscribeChannel` 这个消息通道所使用。广播消息分发器会把消息分发给所有的 `MessageHandler`：

SubscribableChannel messageChannel = new PublishSubscribeChannel();
    
    messageChannel.subscribe(msg -> {
      System.out.println("receive1: " + msg.getPayload());
    });
    
    messageChannel.subscribe(msg -> {
      System.out.println("receive2: " + msg.getPayload());
    });
    
    messageChannel.send(MessageBuilder.withPayload("msg from alibaba").build());
    messageChannel.send(MessageBuilder.withPayload("msg from alibaba").build());

发送两个消息，都被所有的 `MessageHandler` 所消费。控制台打印：

receive1: msg from alibaba
    receive2: msg from alibaba
    receive1: msg from alibaba
    receive2: msg from alibaba

### Spring Cloud Stream ###

SCS与各模块之间的关系是：

*  SCS 在 Spring Integration 的基础上进行了封装，提出了 `Binder`, `Binding`, `@EnableBinding`, `@StreamListener` 等概念;
 *  SCS 与 Spring Boot Actuator 整合，提供了 `/bindings`, `/channels` endpoint;
 *  SCS 与 Spring Boot Externalized Configuration 整合，提供了 `BindingProperties`, `BinderProperties` 等外部化配置类;
 *  SCS 增强了消息发送失败的和消费失败情况下的处理逻辑等功能。
 *  SCS 是 Spring Integration 的加强，同时与 Spring Boot 体系进行了融合，也是 Spring Cloud Bus 的基础。它屏蔽了底层消息中间件的实现细节，希望以统一的一套 API 来进行消息的发送/消费，底层消息中间件的实现细节由各消息中间件的 Binder 完成。

`Binder` 是提供与外部消息中间件集成的组件，为构造 `Binding`提供了 2 个方法，分别是 `bindConsumer` 和 `bindProducer` ，它们分别用于构造生产者和消费者。目前官方的实现有 [Rabbit Binder][] 和 [Kafka Binder][]， [Spring Cloud Alibaba][] 内部已经实现了 [RocketMQ Binder][]。

![fa3476760d50f1213186cf829d35fb4331c.jpg][]

从图中可以看出，`Binding` 是连接应用程序跟消息中间件的桥梁，用于消息的消费和生产。我们来看一个最简单的使用 RocketMQ Binder 的例子，然后分析一下它的底层处理原理：

*  启动类及消息的发送：

@SpringBootApplication
    @EnableBinding({ Source.class, Sink.class }) // 1
    public class SendAndReceiveApplication {
      
        public static void main(String[] args) {
            SpringApplication.run(SendAndReceiveApplication.class, args);
        }
      
        @Bean // 2
        public CustomRunner customRunner() {
            return new CustomRunner();
        }
    
        public static class CustomRunner implements CommandLineRunner {
    
            @Autowired
            private Source source;
    
            @Override
            public void run(String... args) throws Exception {
                int count = 5;
                for (int index = 1; index <= count; index++) {
                    source.output().send(MessageBuilder.withPayload("msg-" + index).build()); // 3
                }
            }
        }
    }

*  消息的接收：

@Service
    public class StreamListenerReceiveService {
    
        @StreamListener(Sink.INPUT) // 4
        public void receiveByStreamListener1(String receiveMsg) {
            System.out.println("receiveByStreamListener: " + receiveMsg);
        }
    
    }

这段代码很简单，没有涉及到 RocketMQ 相关的代码，消息的发送和接收都是基于 SCS 体系完成的。如果想切换成 RabbitMQ 或 kafka，只需修改配置文件即可，代码无需修改。

我们分析这段代码的原理:

1.  `@EnableBinding` 对应的两个接口属性 `Source` 和 `Sink` 是 SCS 内部提供的。SCS 内部会基于 `Source` 和 `Sink` 构造 `BindableProxyFactory`，且对应的 output 和 input 方法返回的 MessageChannel 是 `DirectChannel`。output 和 input 方法修饰的注解对应的 value 是配置文件中 binding 的 name。

public interface Source {
        String OUTPUT = "output";
        @Output(Source.OUTPUT)
        MessageChannel output();
    }
    public interface Sink {
        String INPUT = "input";
        @Input(Sink.INPUT)
        SubscribableChannel input();
    }

配置文件里 bindings 的 name 为 output 和 input，对应 `Source` 和 `Sink` 接口的方法上的注解里的 value：

spring.cloud.stream.bindings.output.destination=test-topic
    spring.cloud.stream.bindings.output.content-type=text/plain
    spring.cloud.stream.rocketmq.bindings.output.producer.group=demo-group
    
    spring.cloud.stream.bindings.input.destination=test-topic
    spring.cloud.stream.bindings.input.content-type=text/plain
    spring.cloud.stream.bindings.input.group=test-group1

1.  构造 `CommandLineRunner`，程序启动的时候会执行 `CustomRunner` 的 `run` 方法。
2.  调用 `Source` 接口里的 output 方法获取 `DirectChannel`，并发送消息到这个消息通道中。这里跟之前 Spring Integration 章节里的代码一致。

*  Source 里的 output 发送消息到 `DirectChannel` 消息通道之后会被 `AbstractMessageChannelBinder#SendingHandler` 这个 `MessageHandler` 处理，然后它会委托给 `AbstractMessageChannelBinder#createProducerMessageHandler` 创建的 MessageHandler 处理(该方法由不同的消息中间件实现)；
 *  不同的消息中间件对应的 `AbstractMessageChannelBinder#createProducerMessageHandler` 方法返回的 MessageHandler 内部会把 Spring Message 转换成对应中间件的 Message 模型并发送到对应中间件的 broker；

1.  使用 `@StreamListener` 进行消息的订阅。请注意，注解里的 `Sink.input` 对应的值是 "input"，会根据配置文件里 binding 对应的 name 为 input 的值进行配置：

*  不同的消息中间件对应的 `AbstractMessageChannelBinder#createConsumerEndpoint` 方法会使用 Consumer 订阅消息，订阅到消息后内部会把中间件对应的 Message 模型转换成 Spring Message；
 *  消息转换之后会把 Spring Message 发送至 name 为 input 的消息通道中；
 *  `@StreamListener` 对应的 `StreamListenerMessageHandler` 订阅了 name 为 input 的消息通道，进行了消息的消费；

这个过程文字描述有点啰嗦，用一张图总结一下(黄色部分涉及到各消息中间件的 Binder 实现以及 MQ 基本的订阅发布功能)：

![ad585a8a4d75ccdde7949e9dddce5add3af.jpg][]

SCS 章节的最后，我们来看一段 SCS 关于消息的处理方式的一段代码：

@StreamListener(value = Sink.INPUT, condition = "headers['index']=='1'")
    public void receiveByHeader(Message msg) {
      System.out.println("receive by headers['index']=='1': " + msg);
    }
    
    @StreamListener(value = Sink.INPUT, condition = "headers['index']=='9999'")
    public void receivePerson(@Payload Person person) {
      System.out.println("receive Person: " + person);
    }
    
    @StreamListener(value = Sink.INPUT)
    public void receiveAllMsg(String msg) {
      System.out.println("receive allMsg by StreamListener. content: " + msg);
    }
    
    @StreamListener(value = Sink.INPUT)
    public void receiveHeaderAndMsg(@Header("index") String index, Message msg) {
      System.out.println("receive by HeaderAndMsg by StreamListener. content: " + msg);
    }

有没有发现这段代码跟 Spring MVC Controller 中接收请求的代码很像? 实际上他们的架构都是类似的，Spring MVC 对于 Controller 中参数和返回值的处理类分别是 `org.springframework.web.method.support.HandlerMethodArgumentResolver`、 `org.springframework.web.method.support.HandlerMethodReturnValueHandler`。

Spring Messaging 中对于参数和返回值的处理类之前也提到过，分别是 `org.springframework.messaging.handler.invocation.HandlerMethodArgumentResolver`、`org.springframework.messaging.handler.invocation.HandlerMethodReturnValueHandler`。

它们的类名一模一样，甚至内部的方法名也一样。

### 总结 ###

![eb428b1d78eca95ecee4e65df3b8f968f90.jpg][]

上图是 SCS 体系相关类说明的总结，关于 SCS 以及 RocketMQ Binder 更多相关的示例，可以参考 [RocketMQ Binder Demos][]，包含了消息的聚合、分割、过滤；消息异常处理；消息标签、sql过滤；同步、异步消费等等。

下一篇文章，我们将分析消息总线(Spring Cloud Bus) 在 Spring Cloud 体系中的作用，并逐步展开，分析 Spring Cloud Alibaba 中的 RocketMQ Binder 是如何实现 Spring Cloud Stream 标准的。

[原文链接][Link 1]  
本文为云栖社区原创内容，未经允许不得转载。

转载于:https://my.oschina.net/u/1464083/blog/3034002

[2019_Python_]: https://my.oschina.net/u/2663968/blog/3061697
[hot3.png]: /images/20220114/f3d4dd93ceed40009bd4f820185fa63e.png
[Spring Cloud Stream]: https://spring.io/projects/spring-cloud-stream
[f17d6b9b23908b3e24798eb0cd4c3b61479.jpg]: /images/20220114/043479a106d348dc88bb2a09c76aa7ab.png
[4beb01d755ca4a3b0b3cf224259f3fda00a.jpg]: /images/20220114/a8473928c1e5460cbff8821509804fa8.png
[Spring Integration]: https://docs.spring.io/spring-integration/reference/html/index.html
[Enterprise Integration Patterns]: https://www.enterpriseintegrationpatterns.com/
[a89cb59cd3bf1e211977248597510544cfe.jpg]: /images/20220114/e9964a8fc95d4e0c8694b97284d50f3a.png
[3539199a64eed63ba5a01bb8d3e12ef674b.jpg]: /images/20220114/503468f914fe4d288da393a9ce87d93a.png
[2ae2420ec57395d2d617b8e3526f5dd2f76.jpg]: /images/20220114/b8c48339310e4f69a3f36c5361798491.png
[58f4e010cf3b8eaef2853acd584563104d6.jpg]: /images/20220114/8fad91814c5d4c24a0dc71c54fee8164.png
[Rabbit Binder]: https://github.com/spring-cloud/spring-cloud-stream-binder-rabbit
[Kafka Binder]: https://github.com/spring-cloud/spring-cloud-stream-binder-kafka
[Spring Cloud Alibaba]: https://github.com/spring-cloud-incubator/spring-cloud-alibaba
[RocketMQ Binder]: https://github.com/spring-cloud-incubator/spring-cloud-alibaba/tree/master/spring-cloud-stream-binder-rocketmq
[fa3476760d50f1213186cf829d35fb4331c.jpg]: /images/20220114/a7f9a29b3b844ac093efe7b78ccde10e.png
[ad585a8a4d75ccdde7949e9dddce5add3af.jpg]: /images/20220114/bca871ba5a75416aa28cce9fd1c4aec9.png
[eb428b1d78eca95ecee4e65df3b8f968f90.jpg]: /images/20220114/0b5f9e9f066c448fa3f99bf7387c73ca.png
[RocketMQ Binder Demos]: https://github.com/fangjian0423/rocketmq-binder-demo
[Link 1]: https://yq.aliyun.com/articles/696830?utm_content=g_1000051114