Spring Cloud Stream 體系及原理介紹

簡介： Spring Cloud Stream在 Spring Cloud 體系內用於構建高度可擴展的基於事件驅動的微服務，其目的是爲了簡化消息在 Spring Cloud 應用程序中的開發。

做者 | 洛夜
來源 | 阿里巴巴雲原生公衆號

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 章節裏的代碼一致。
3. Source 裏的 output 發送消息到DirectChannel消息通道以後會被AbstractMessageChannelBinder#SendingHandler這個MessageHandler處理，而後它會委託給AbstractMessageChannelBinder#createProducerMessageHandler建立的 MessageHandler 處理(該方法由不一樣的消息中間件實現)。
4. 不一樣的消息中間件對應的AbstractMessageChannelBinder#createProducerMessageHandler方法返回的 MessageHandler 內部會把 Spring Message 轉換成對應中間件的 Message 模型併發送到對應中間件的 broker。
5. 使用@StreamListener進行消息的訂閱。請注意，註解裏的Sink.input對應的值是 "input"，會根據配置文件裏 binding 對應的 name 爲 input 的值進行配置：
6. 不一樣的消息中間件對應的AbstractMessageChannelBinder#createConsumerEndpoint方法會使用 Consumer 訂閱消息，訂閱到消息後內部會把中間件對應的 Message 模型轉換成 Spring Message。
7. 消息轉換以後會把 Spring Message 發送至 name 爲 input 的消息通道中。
8. @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 過濾；同步、異步消費等等。
原文連接本文爲阿里雲原創內容，未經容許不得轉載。