使用 Spring Cloud Stream 構建消息驅動微服務

 

相關源碼： spring cloud demo

微服務的目的： 松耦合

事件驅動的優勢：高度解耦

Spring Cloud Stream 的幾個概念

Spring Cloud Stream is a framework for building message-driven microservice applications.

官方定義 Spring Cloud Stream 是一個構建消息驅動微服務的框架。

Spring Cloud Stream Application

應用程序通過 inputs 或者 outputs 來與 Spring Cloud Stream 中binder 交互，通過我們配置來 binding ，而 Spring Cloud Stream 的 binder 負責與中間件交互。所以，我們只需要搞清楚如何與 Spring Cloud Stream 交互就可以方便使用消息驅動的方式

Binder

Binder 是 Spring Cloud Stream 的一個抽象概念，是應用與消息中間件之間的粘合劑。目前 Spring Cloud Stream 實現了 Kafka 和 Rabbit MQ 的binder。

通過 binder ，可以很方便的連接中間件，可以動態的改變消息的
destinations（對應於 Kafka 的topic，Rabbit MQ 的 exchanges），這些都可以通過外部配置項來做到。

甚至可以任意的改變中間件的類型而不需要修改一行代碼。

Publish-Subscribe

消息的發布（Publish）和訂閱（Subscribe）是事件驅動的經典模式。Spring Cloud Stream 的數據交互也是基於這個思想。生產者把消息通過某個 topic 廣播出去（Spring Cloud Stream 中的 destinations）。其他的微服務，通過訂閱特定 topic 來獲取廣播出來的消息來觸發業務的進行。

這種模式，極大的降低了生產者與消費者之間的耦合。即使有新的應用的引入，也不需要破壞當前系統的整體結構。

Consumer Groups

“Group”，如果使用過 Kafka 的童鞋並不會陌生。Spring Cloud Stream 的這個分組概念的意思基本和 Kafka 一致。

微服務中動態的縮放同一個應用的數量以此來達到更高的處理能力是非常必須的。對於這種情況，同一個事件防止被重復消費，只要把這些應用放置於同一個 “group” 中，就能夠保證消息只會被其中一個應用消費一次。

Durability

消息事件的持久化是必不可少的。Spring Cloud Stream 可以動態的選擇一個消息隊列是持久化，還是 present。

Bindings

bindings 是我們通過配置把應用和spring cloud stream 的 binder 綁定在一起，之后我們只需要修改 binding 的配置來達到動態修改topic、exchange、type等一系列信息而不需要修改一行代碼。

基於 RabbitMQ 使用

以下內容源碼： spring cloud demo

消息接收

Spring Cloud Stream 基本用法，需要定義一個接口，如下是內置的一個接口。

public interface Sink {
    String INPUT = "input";

    @Input("input")
    SubscribableChannel input();
}

注釋 @Input 對應的方法，需要返回 SubscribableChannel ，並且參入一個參數值。

這就接口聲明了一個 binding 命名為 “input” 。

其他內容通過配置指定：

spring:
  cloud:
    stream:
      bindings:
        input:
          destination: mqTestDefault

destination：指定了消息獲取的目的地，對應於MQ就是 exchange，這里的exchange就是 mqTestDefault

@SpringBootApplication
@EnableBinding(Sink.class)
public class Application {

    // 監聽 binding 為 Sink.INPUT 的消息
    @StreamListener(Sink.INPUT)
    public void input(Message<String> message) {
        System.out.println("一般監聽收到：" + message.getPayload());
    }

    public static void main(String[] args) {
        SpringApplication.run(Application.class);
    }
}

定義一個 class （這里直接在啟動類），並且添加注解@EnableBinding(Sink.class) ，其中 Sink 就是上述的接口。同時定義一個方法（此處是 input）標明注解為@StreamListener(Processor.INPUT) ，方法參數為 Message 。

啟動后，默認是會創建一個臨時隊列，臨時隊列綁定的exchange為 “mqTestDefault”，routing key為 “#”。

所有發送 exchange 為“mqTestDefault” 的MQ消息都會被投遞到這個臨時隊列，並且觸發上述的方法。

以上代碼就完成了最基本的消費者部分。

消息發送

消息的發送同消息的接受，都需要定義一個接口，不同的是接口方法的返回對象是MessageChannel，下面是 Spring Cloud Stream 內置的接口：

public interface Source {
    String OUTPUT = "output";

    @Output("output")
    MessageChannel output();
}

這就接口聲明了一個 binding 命名為 “output” ，不同於上述的 “input”，這個binding 聲明了一個消息輸出流，也就是消息的生產者。

spring:
  cloud:
    stream:
      bindings:
        output:
          destination: mqTestDefault
          contentType: text/plain

contentType：用於指定消息的類型。具體可以參考 spring cloud stream docs

destination：指定了消息發送的目的地，對應 RabbitMQ，會發送到 exchange 是 mqTestDefault 的所有消息隊列中。

代碼中調用：

@SpringBootApplication
@EnableBinding(Source.class)
public class Application implements CommandLineRunner {

    @Autowired
    @Qualifier("output")
    MessageChannel output;

    @Override
    public void run(String... strings) throws Exception {
        // 字符串類型發送MQ
        System.out.println("字符串信息發送");
        output.send(MessageBuilder.withPayload("大家好").build());
    }

    public static void main(String[] args) {
        SpringApplication.run(Application.class);
    }

}

通過注入MessageChannel的方式，發送消息。

通過注入Source 接口的方式，發送消息。 具體可以查看樣例

以上代碼就完成了最基本的生產者部分。

自定義消息發送接收

自定義接口

Spring Cloud Stream 內置了兩種接口，分別定義了 binding 為 “input” 的輸入流，和 “output” 的輸出流，而在我們實際使用中，往往是需要定義各種輸入輸出流。使用方法也很簡單。

interface OrderProcessor {

    String INPUT_ORDER = "inputOrder";
    String OUTPUT_ORDER = "outputOrder";

    @Input(INPUT_ORDER)
    SubscribableChannel inputOrder();

    @Output(OUTPUT_ORDER)
    MessageChannel outputOrder();
}

一個接口中，可以定義無數個輸入輸出流，可以根據實際業務情況划分。上述的接口，定義了一個訂單輸入，和訂單輸出兩個 binding。

使用時，需要在 @EnableBinding 注解中，添加自定義的接口。
使用 @StreamListener 做監聽的時候，需要指定OrderProcessor.INPUT_ORDER

spring:
  cloud:
    stream:
      defaultBinder: defaultRabbit
      bindings:
        inputOrder:
          destination: mqTestOrder
        outputOrder:
          destination: mqTestOrder

如上配置，指定了 destination 為 mqTestOrder 的輸入輸出流。

分組與持久化

上述自定義的接口配置中，Spring Cloud Stream 會在 RabbitMQ 中創建一個臨時的隊列，程序關閉，對應的連接關閉的時候，該隊列也會消失。而在實際使用中，我們需要一個持久化的隊列，並且指定一個分組，用於保證應用服務的縮放。

只需要在消費者端的 binding 添加配置項 spring.cloud.stream.bindings.[channelName].group = XXX 。對應的隊列就是持久化，並且名稱為：mqTestOrder.XXX。

rabbitMQ routing key 綁定

用慣了 rabbitMQ 的童鞋，在使用的時候，發現 Spring Cloud Stream 的消息投遞，默認是根據 destination + group 進行區分，所有的消息都投遞到 routing key 為 “#‘’ 的消息隊列里。

如果我們需要進一步根據 routing key 來進行區分消息投遞的目的地，或者消息接受，需要進一步配，Spring Cloud Stream 也提供了相關配置：

spring:
  cloud:
    stream:
      bindings:
        inputProductAdd:
          destination: mqTestProduct
          group: addProductHandler      # 擁有 group 默認會持久化隊列
        outputProductAdd:
          destination: mqTestProduct
      rabbit:
        bindings:
          inputProductAdd:
            consumer:
              bindingRoutingKey: addProduct.*       # 用來綁定消費者的 routing key
          outputProductAdd:
            producer:
              routing-key-expression: '''addProduct.*'''  # 需要用這個來指定 RoutingKey

spring.cloud.stream.rabbit.bindings.[channelName].consumer.bindingRoutingKey
指定了生成的消息隊列的routing key

spring.cloud.stream.rabbit.bindings.[channelName].producer.routing-key-expression 指定了生產者消息投遞的routing key

DLX 隊列

DLX 作用

DLX:Dead-Letter-Exchange（死信隊列）。利用DLX, 當消息在一個隊列中變成死信（dead message）之后，它能被重新publish到另一個Exchange，這個Exchange就是DLX。消息變成死信一向有一下幾種情況：

消息被拒絕（basic.reject/ basic.nack）並且requeue=false
消息TTL過期（參考：RabbitMQ之TTL（Time-To-Live 過期時間））
隊列達到最大長度

DLX也是一個正常的Exchange，和一般的Exchange沒有區別，它能在任何的隊列上被指定，實際上就是設置某個隊列的屬性，當這個隊列中有死信時，RabbitMQ就會自動的將這個消息重新發布到設置的Exchange上去，進而被路由到另一個隊列，可以監聽這個隊列中消息做相應的處理。

Spring Cloud Stream 中使用

spring.cloud.stream.rabbit.bindings.[channelName].consumer.autoBindDlq=true

spring.cloud.stream.rabbit.bindings.[channelName].consumer.republishToDlq=true

配置說明，可以參考 spring cloud stream rabbitmq consumer properties

結論

Spring Cloud Stream 最大的方便之處，莫過於抽象了事件驅動的一些概念，對於消息中間件的進一步封裝，可以做到代碼層面對中間件的無感知，甚至於動態的切換中間件，切換topic。使得微服務開發的高度解耦，服務可以關注更多自己的業務流程。

相關文檔

spring cloud stream 文檔

spring cloud stream 項目

spring cloud stream 樣例