作者 | 洛夜
起源 | 阿里巴巴云原生公众号

Spring Cloud Stream在 Spring Cloud 体系内用于构建高度可扩大的基于事件驱动的微服务，其目标是为了简化音讯在 Spring Cloud 应用程序中的开发。

Spring Cloud Stream (前面以 SCS 代替 Spring Cloud Stream) 自身内容很多，而且它还有很多内部的依赖，想要相熟 SCS，必须要先理解 Spring Messaging 和 Spring Integration 这两个我的项目，接下来，文章将围绕以下三点进行开展：

• 什么是 Spring Messaging
• 什么是 Spring Integration
• 什么是 SCS 体系及其原理

本文配套可交互教程已登录阿里云知口头手实验室，PC 端登录 start.aliyun.com_ _在浏览器中立刻体验。

Spring Messaging

Spring Messaging 是 Spring Framework 中的一个模块，其作用就是对立音讯的编程模型。

• 比方音讯 Messaging 对应的模型就包含一个音讯体 Payload 和音讯头 Header：

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

• 音讯通道 MessageChannel 用于接管音讯，调用 send 办法能够将音讯发送至该音讯通道中：

@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 外部在音讯模型的根底上衍生出了其它的一些性能，如：

• 音讯接管参数及返回值解决：音讯接管参数处理器 HandlerMethodArgumentResolver 配合 @Header, @Payload 等注解应用；音讯接管后的返回值处理器 HandlerMethodReturnValueHandler 配合 @SendTo 注解应用；
• 音讯体内容转换器 MessageConverter；
• 对立形象的音讯发送模板 AbstractMessageSendingTemplate；
• 音讯通道拦截器 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 等内容。

这里为大家介绍几种音讯的解决形式：

• 音讯的宰割：

• 音讯的聚合：

• 音讯的过滤：

• 音讯的散发：

接下来，咱们以一个最简略的例子来尝试一下 Spring Integration。

这段代码解释为：

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

• 结构一个可订阅的音讯通道 messageChannel。
• 应用 MessageHandler 去生产这个音讯通道里的音讯。
• 发送一条音讯到这个音讯通道，音讯最终被音讯通道里的 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());

Spring Cloud Stream

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

• SCS 在 Spring Integration 的根底上进行了封装，提出了 Binder, Binding, @EnableBinding, @StreamListener 等概念。
• SCS 与 Spring Boot Actuator 整合，提供了 /bindings, /channelsendpoint。
• 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。

从图中能够看出，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 根本的订阅公布性能)：

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。

它们的类名截然不同，甚至外部的办法名也一样。

总结

上图是 SCS 体系相干类阐明的总结，对于 SCS 以及 RocketMQ Binder 更多相干的示例，能够参考 RocketMQ Binder Demos，蕴含了音讯的聚合、宰割、过滤；音讯异样解决；音讯标签、SQL 过滤；同步、异步生产等等。