相關(guān)源碼： spring cloud demo

微服務(wù)的目的： 松耦合

事件驅(qū)動(dòng)的優(yōu)勢(shì)：高度解耦

Spring Cloud Stream 的幾個(gè)概念

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

官方定義 Spring Cloud Stream 是一個(gè)構(gòu)建消息驅(qū)動(dòng)微服務(wù)的框架。

Spring Cloud Stream Application

應(yīng)用程序通過(guò) inputs 或者 outputs 來(lái)與 Spring Cloud Stream 中binder 交互，通過(guò)我們配置來(lái) binding ，而 Spring Cloud Stream 的 binder 負(fù)責(zé)與中間件交互。所以，我們只需要搞清楚如何與 Spring Cloud Stream 交互就可以方便使用消息驅(qū)動(dòng)的方式

Binder

Binder 是 Spring Cloud Stream 的一個(gè)抽象概念，是應(yīng)用與消息中間件之間的粘合劑。目前 Spring Cloud Stream 實(shí)現(xiàn)了 Kafka 和 Rabbit MQ 的binder。

通過(guò) binder ，可以很方便的連接中間件，可以動(dòng)態(tài)的改變消息的
destinations（對(duì)應(yīng)于 Kafka 的topic，Rabbit MQ 的 exchanges），這些都可以通過(guò)外部配置項(xiàng)來(lái)做到。

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

Publish-Subscribe

消息的發(fā)布（Publish）和訂閱（Subscribe）是事件驅(qū)動(dòng)的經(jīng)典模式。Spring Cloud Stream 的數(shù)據(jù)交互也是基于這個(gè)思想。生產(chǎn)者把消息通過(guò)某個(gè) topic 廣播出去（Spring Cloud Stream 中的 destinations）。其他的微服務(wù)，通過(guò)訂閱特定 topic 來(lái)獲取廣播出來(lái)的消息來(lái)觸發(fā)業(yè)務(wù)的進(jìn)行。

這種模式，極大的降低了生產(chǎn)者與消費(fèi)者之間的耦合。即使有新的應(yīng)用的引入，也不需要破壞當(dāng)前系統(tǒng)的整體結(jié)構(gòu)。

Consumer Groups

“Group”，如果使用過(guò) Kafka 的童鞋并不會(huì)陌生。Spring Cloud Stream 的這個(gè)分組概念的意思基本和 Kafka 一致。

微服務(wù)中動(dòng)態(tài)的縮放同一個(gè)應(yīng)用的數(shù)量以此來(lái)達(dá)到更高的處理能力是非常必須的。對(duì)于這種情況，同一個(gè)事件防止被重復(fù)消費(fèi)，只要把這些應(yīng)用放置于同一個(gè) “group” 中，就能夠保證消息只會(huì)被其中一個(gè)應(yīng)用消費(fèi)一次。

Durability

消息事件的持久化是必不可少的。Spring Cloud Stream 可以動(dòng)態(tài)的選擇一個(gè)消息隊(duì)列是持久化，還是 present。

Bindings

bindings 是我們通過(guò)配置把應(yīng)用和spring cloud stream 的 binder 綁定在一起，之后我們只需要修改 binding 的配置來(lái)達(dá)到動(dòng)態(tài)修改topic、exchange、type等一系列信息而不需要修改一行代碼。

基于 RabbitMQ 使用

以下內(nèi)容源碼： spring cloud demo

消息接收

Spring Cloud Stream 基本用法，需要定義一個(gè)接口，如下是內(nèi)置的一個(gè)接口。

public interface Sink {
    String INPUT = "input";

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

注釋__ @Input__ 對(duì)應(yīng)的方法，需要返回 __ SubscribableChannel __ ，并且參入一個(gè)參數(shù)值。

這就接口聲明了一個(gè)__ binding __命名為 “input” 。

其他內(nèi)容通過(guò)配置指定：

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

destination：指定了消息獲取的目的地，對(duì)應(yīng)于MQ就是 exchange，這里的exchange就是 mqTestDefault

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

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

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

定義一個(gè) class （這里直接在啟動(dòng)類(lèi)），并且添加注解@EnableBinding(Sink.class) ，其中 Sink 就是上述的接口。同時(shí)定義一個(gè)方法（此處是 input）標(biāo)明注解為 __ @StreamListener(Processor.INPUT) __，方法參數(shù)為 Message 。

啟動(dòng)后，默認(rèn)是會(huì)創(chuàng)建一個(gè)臨時(shí)隊(duì)列，臨時(shí)隊(duì)列綁定的exchange為 “mqTestDefault”，routing key為 “#”。

所有發(fā)送 exchange 為“mqTestDefault” 的MQ消息都會(huì)被投遞到這個(gè)臨時(shí)隊(duì)列，并且觸發(fā)上述的方法。

以上代碼就完成了最基本的消費(fèi)者部分。

消息發(fā)送

消息的發(fā)送同消息的接受，都需要定義一個(gè)接口，不同的是接口方法的返回對(duì)象是 MessageChannel，下面是 Spring Cloud Stream 內(nèi)置的接口：

public interface Source {
    String OUTPUT = "output";

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

這就接口聲明了一個(gè) binding 命名為 “output” ，不同于上述的 “input”，這個(gè)binding 聲明了一個(gè)消息輸出流，也就是消息的生產(chǎn)者。

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

contentType：用于指定消息的類(lèi)型。具體可以參考 spring cloud stream docs

destination：指定了消息發(fā)送的目的地，對(duì)應(yīng) RabbitMQ，會(huì)發(fā)送到 exchange 是 mqTestDefault 的所有消息隊(duì)列中。

代碼中調(diào)用：

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

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

    @Override
    public void run(String... strings) throws Exception {
        // 字符串類(lèi)型發(fā)送MQ
        System.out.println("字符串信息發(fā)送");
        output.send(MessageBuilder.withPayload("大家好").build());
    }
    
    public static void main(String[] args) {
        SpringApplication.run(Application.class);
    }
    
}

通過(guò)注入MessageChannel的方式，發(fā)送消息。

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

以上代碼就完成了最基本的生產(chǎn)者部分。

自定義消息發(fā)送接收

自定義接口

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

interface OrderProcessor {

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

    @Input(INPUT_ORDER)
    SubscribableChannel inputOrder();

    @Output(OUTPUT_ORDER)
    MessageChannel outputOrder();
}

一個(gè)接口中，可以定義無(wú)數(shù)個(gè)輸入輸出流，可以根據(jù)實(shí)際業(yè)務(wù)情況劃分。上述的接口，定義了一個(gè)訂單輸入，和訂單輸出兩個(gè) binding。

使用時(shí)，需要在 @EnableBinding 注解中，添加自定義的接口。
使用 @StreamListener 做監(jiān)聽(tīng)的時(shí)候，需要指定 OrderProcessor.INPUT_ORDER

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

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

分組與持久化

上述自定義的接口配置中，Spring Cloud Stream 會(huì)在 RabbitMQ 中創(chuàng)建一個(gè)臨時(shí)的隊(duì)列，程序關(guān)閉，對(duì)應(yīng)的連接關(guān)閉的時(shí)候，該隊(duì)列也會(huì)消失。而在實(shí)際使用中，我們需要一個(gè)持久化的隊(duì)列，并且指定一個(gè)分組，用于保證應(yīng)用服務(wù)的縮放。

只需要在消費(fèi)者端的 binding 添加配置項(xiàng) spring.cloud.stream.bindings.[channelName].group = XXX 。對(duì)應(yīng)的隊(duì)列就是持久化，并且名稱(chēng)為：mqTestOrder.XXX。

rabbitMQ routing key 綁定

用慣了 rabbitMQ 的童鞋，在使用的時(shí)候，發(fā)現(xiàn) Spring Cloud Stream 的消息投遞，默認(rèn)是根據(jù) destination + group 進(jìn)行區(qū)分，所有的消息都投遞到 routing key 為 “#‘’ 的消息隊(duì)列里。

如果我們需要進(jìn)一步根據(jù) routing key 來(lái)進(jìn)行區(qū)分消息投遞的目的地，或者消息接受，需要進(jìn)一步配，Spring Cloud Stream 也提供了相關(guān)配置：

spring:
  cloud:
    stream:
      bindings:
        inputProductAdd:
          destination: mqTestProduct
          group: addProductHandler      # 擁有 group 默認(rèn)會(huì)持久化隊(duì)列
        outputProductAdd:
          destination: mqTestProduct
      rabbit:
        bindings:
          inputProductAdd:
            consumer:
              bindingRoutingKey: addProduct.*       # 用來(lái)綁定消費(fèi)者的 routing key
          outputProductAdd:
            producer:
              routing-key-expression: '''addProduct.*'''  # 需要用這個(gè)來(lái)指定 RoutingKey

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

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

DLX 隊(duì)列

DLX 作用

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

消息被拒絕（basic.reject/ basic.nack）并且requeue=false
消息TTL過(guò)期（參考：RabbitMQ之TTL（Time-To-Live 過(guò)期時(shí)間））
隊(duì)列達(dá)到最大長(zhǎng)度

DLX也是一個(gè)正常的Exchange，和一般的Exchange沒(méi)有區(qū)別，它能在任何的隊(duì)列上被指定，實(shí)際上就是設(shè)置某個(gè)隊(duì)列的屬性，當(dāng)這個(gè)隊(duì)列中有死信時(shí)，RabbitMQ就會(huì)自動(dòng)的將這個(gè)消息重新發(fā)布到設(shè)置的Exchange上去，進(jìn)而被路由到另一個(gè)隊(duì)列，可以監(jiān)聽(tīng)這個(gè)隊(duì)列中消息做相應(yīng)的處理。

Spring Cloud Stream 中使用

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

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

配置說(shuō)明，可以參考 spring cloud stream rabbitmq consumer properties

結(jié)論

Spring Cloud Stream 最大的方便之處，莫過(guò)于抽象了事件驅(qū)動(dòng)的一些概念，對(duì)于消息中間件的進(jìn)一步封裝，可以做到代碼層面對(duì)中間件的無(wú)感知，甚至于動(dòng)態(tài)的切換中間件，切換topic。使得微服務(wù)開(kāi)發(fā)的高度解耦，服務(wù)可以關(guān)注更多自己的業(yè)務(wù)流程。

相關(guān)文檔

spring cloud stream 文檔

spring cloud stream 項(xiàng)目

spring cloud stream 樣例