之前在寫Spring Boot基礎教程的時候寫過一篇《Spring Boot中使用RabbitMQ》。在該文中癣籽，我們通過簡單的配置和注解就能實現(xiàn)向RabbitMQ中生產(chǎn)和消費消息杨帽。實際上我們使用的對RabbitMQ的starter就是通過Spring Cloud Stream中對RabbitMQ的支持來實現(xiàn)的酬荞。下面我們就通過本文來了解一下Spring Cloud Stream嵌赠。

Spring Cloud Stream是一個用來為微服務應用構建消息驅動能力的框架板甘。它可以基于Spring Boot來創(chuàng)建獨立的叫榕、可用于生產(chǎn)的Spring應用程序浑侥。它通過使用Spring Integration來連接消息代理中間件以實現(xiàn)消息事件驅動的微服務應用。Spring Cloud Stream為一些供應商的消息中間件產(chǎn)品提供了個性化的自動化配置實現(xiàn)晰绎，并且引入了發(fā)布-訂閱寓落、消費組以及消息分區(qū)這三個核心概念。簡單的說荞下，Spring Cloud Stream本質上就是整合了Spring Boot和Spring Integration伶选，實現(xiàn)了一套輕量級的消息驅動的微服務框架史飞。通過使用Spring Cloud Stream，可以有效地簡化開發(fā)人員對消息中間件的使用復雜度仰税，讓系統(tǒng)開發(fā)人員可以有更多的精力關注于核心業(yè)務邏輯的處理构资。由于Spring Cloud Stream基于Spring Boot實現(xiàn)，所以它秉承了Spring Boot的優(yōu)點陨簇，實現(xiàn)了自動化配置的功能幫忙我們可以快速的上手使用吐绵，但是目前為止Spring Cloud Stream只支持下面兩個著名的消息中間件的自動化配置：

• RabbitMQ
• Kafka

快速入門

下面我們通過構建一個簡單的示例來對Spring Cloud Stream有一個初步認識。該示例主要目標是構建一個基于Spring Boot的微服務應用河绽，這個微服務應用將通過使用消息中間件RabbitMQ來接收消息并將消息打印到日志中己单。所以，在進行下面步驟之前請先確認已經(jīng)在本地安裝了RabbitMQ耙饰，具體安裝步驟請參考此文荷鼠。

構建一個Spring Cloud Stream消費者

• 創(chuàng)建一個基礎的Spring Boot工程，命名為：stream-hello

• 編輯pom.xml中的依賴關系榔幸，引入Spring Cloud Stream對RabbitMQ的支持允乐，具體如下：

<parent>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-parent</artifactId>
    <version>1.5.9.RELEASE</version>
    <relativePath/> <!-- lookup parent from repository -->
</parent>

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>

    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-starter-stream-rabbit</artifactId>     
    </dependency>
</dependencies>

<dependencyManagement>
    <dependencies>
        <dependency>
            <groupId>org.springframework.cloud</groupId>
            <artifactId>spring-cloud-dependencies</artifactId>
            <version>Dalston.SR4</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>

• 創(chuàng)建用于接收來自RabbitMQ消息的消費者SinkReceiver，具體如下：

@EnableBinding(Sink.class)
public class SinkReceiver {

    private static Logger logger = LoggerFactory.getLogger(SinkReceiver.class);

    @StreamListener(Sink.INPUT)
    public void receive(Object payload) {
        logger.info("Received: " + payload);
    }

}

• 創(chuàng)建應用主類削咆，這里同其他Spring Boot一樣牍疏，沒有什么特別之處，具體如下：

@SpringBootApplication
public class SinkApplication {

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

}

到這里拨齐，我們快速入門示例的編碼任務就已經(jīng)完成了鳞陨。下面我們分別啟動RabbitMQ以及該Spring Boot應用，然后做下面的試驗瞻惋，看看它們是如何運作的厦滤。

手工測試驗證

• 我們先來看一下Spring Boot應用的啟動日志。

...
INFO 16272 --- [main] o.s.c.s.b.r.RabbitMessageChannelBinder   : declaring queue for inbound: input.anonymous.Y8VsFILmSC27eS5StsXp6A, bound to: input
INFO 16272 --- [main] o.s.a.r.c.CachingConnectionFactory       : Created new connection: SimpleConnection@3c78e551 [delegate=amqp://guest@127.0.0.1:5672/]
INFO 16272 --- [main] o.s.integration.channel.DirectChannel    : Channel 'input.anonymous.Y8VsFILmSC27eS5StsXp6A.bridge' has 1 subscriber(s).
INFO 16272 --- [main] o.s.i.a.i.AmqpInboundChannelAdapter      : started inbound.input.anonymous.Y8VsFILmSC27eS5StsXp6A
...

從上面的日志內容中歼狼，我們可以獲得以下信息：

• 使用guest用戶創(chuàng)建了一個指向127.0.0.1:5672位置的RabbitMQ連接掏导，在RabbitMQ的控制臺中我們也可以發(fā)現(xiàn)它。

[圖片上傳失敗...(image-e63c2a-1513694844373)]

• 聲明了一個名為input.anonymous.Y8VsFILmSC27eS5StsXp6A的隊列羽峰，并通過RabbitMessageChannelBinder將自己綁定為它的消費者趟咆。這些信息我們也能在RabbitMQ的控制臺中發(fā)現(xiàn)它們。

[圖片上傳失敗...(image-725d48-1513694844373)]

下面我們可以在RabbitMQ的控制臺中進入input.anonymous.Y8VsFILmSC27eS5StsXp6A隊列的管理頁面，通過Publish Message功能來發(fā)送一條消息到該隊列中。

[圖片上傳失敗...(image-a5cfda-1513694844373)]

此時趋急，我們可以在當前啟動的Spring Boot應用程序的控制臺中看到下面的內容：

INFO 16272 --- [C27eS5StsXp6A-1] com.didispace.HelloApplication           : Received: [B@7cba610e

我們可以發(fā)現(xiàn)在應用控制臺中輸出的內容就是SinkReceiver中receive方法定義的，而輸出的具體內容則是來自消息隊列中獲取的對象幕庐。這里由于我們沒有對消息進行序列化，所以輸出的只是該對象的引用，在后面的小節(jié)中我們會詳細介紹接收消息后的處理。

在順利完成上面快速入門的示例后疆偿，我們簡單解釋一下上面的步驟是如何將我們的Spring Boot應用連接上RabbitMQ來消費消息以實現(xiàn)消息驅動業(yè)務邏輯的妈橄。

首先，我們對Spring Boot應用做的就是引入spring-cloud-starter-stream-rabbit依賴翁脆，該依賴包是Spring Cloud Stream對RabbitMQ支持的封裝，其中包含了對RabbitMQ的自動化配置等內容鼻种。從下面它定義的依賴關系中反番，我們還可以知道它等價于spring-cloud-stream-binder-rabbit依賴。

<dependencies>
    <dependency>
        <groupId>org.springframework.cloud</groupId>
        <artifactId>spring-cloud-stream-binder-rabbit</artifactId>
    </dependency>
</dependencies>

接著叉钥，我們再來看看這里用到的幾個Spring Cloud Stream的核心注解罢缸，它們都被定義在SinkReceiver中：

• @EnableBinding，該注解用來指定一個或多個定義了@Input或@Output注解的接口投队，以此實現(xiàn)對消息通道（Channel）的綁定枫疆。在上面的例子中，我們通過@EnableBinding(Sink.class)綁定了Sink接口敷鸦，該接口是Spring Cloud Stream中默認實現(xiàn)的對輸入消息通道綁定的定義息楔，它的源碼如下：

public interface Sink {

    String INPUT = "input";

    @Input(Sink.INPUT)
    SubscribableChannel input();

}

它通過@Input注解綁定了一個名為input的通道。除了Sink之外扒披，Spring Cloud Stream還默認實現(xiàn)了綁定output通道的Source接口值依，還有結合了Sink和Source的Processor接口，實際使用時我們也可以自己通過@Input和@Output注解來定義綁定消息通道的接口碟案。當我們需要為@EnableBinding指定多個接口來綁定消息通道的時候愿险，可以這樣定義：@EnableBinding(value = {Sink.class, Source.class})。

• @StreamListener：該注解主要定義在方法上价说，作用是將被修飾的方法注冊為消息中間件上數(shù)據(jù)流的事件監(jiān)聽器辆亏，注解中的屬性值對應了監(jiān)聽的消息通道名。在上面的例子中鳖目，我們通過@StreamListener(Sink.INPUT)注解將receive方法注冊為對input消息通道的監(jiān)聽處理器扮叨，所以當我們在RabbitMQ的控制頁面中發(fā)布消息的時候，receive方法會做出對應的響應動作领迈。

編寫消費消息的單元測試用例

上面我們通過RabbitMQ的控制臺完成了發(fā)送消息來驗證了消息消費程序的功能甫匹，雖然這種方法比較low，但是通過上面的步驟惦费，相信大家對RabbitMQ和Spring Cloud Stream的消息消費已經(jīng)有了一些基礎的認識兵迅。下面我們通過編寫生產(chǎn)消息的單元測試用例來完善我們的入門內容。

• 在上面創(chuàng)建的工程中創(chuàng)建單元測試類：

@RunWith(SpringRunner.class)
@EnableBinding(value = {SinkApplicationTests.SinkSender.class})
public class SinkApplicationTests {

    @Autowired
    private SinkSender sinkSender;

    @Test
    public void sinkSenderTester() {
        sinkSender.output().send(MessageBuilder.withPayload("produce a message ：http://blog.didispace.com").build());
    }

    public interface SinkSender {

        String OUTPUT = "input";

        @Output(SinkSender.OUTPUT)
        MessageChannel output();

    }

}

• 在應用了上面的消息消費者程序之后薪贫，運行這里定義的單元測試程序恍箭，我們馬上就能在消息消費者的控制臺中收到下面的內容：

INFO 50947 --- [L2W-c2AcChb2Q-1] com.didispace.stream.SinkReceiver        : Received: produce a message ：http://blog.didispace.com

在上面的單元測試中，我們通過@Output(SinkSender.OUTPUT)定義了一個輸出通過瞧省，而該輸出通道的名稱為input扯夭，與前文中的Sink中定義的消費通道同名鳍贾，所以這里的單元測試與前文的消費者程序組成了一對生產(chǎn)者與消費者。到這里交洗，本文的內容就次結束骑科，如果您能夠獨立的完成上面的例子，那么對于Spring Cloud Stream的基礎使用算是入門了构拳。但是咆爽，Spring Cloud Stream的使用遠不止于此，在近期的博文中置森，我講繼續(xù)更新這部分內容斗埂，幫助他們來理解和用好Spring Cloud Stream來構建消息驅動的微服務！

本文完整實例：

• Github
• Gitee

如果您對這些感興趣凫海，歡迎star呛凶、follow、收藏行贪、轉發(fā)給予支持漾稀！

本文內容部分節(jié)選自我的《Spring Cloud微服務實戰(zhàn)》，但對依賴的Spring Boot和Spring Cloud版本做了升級建瘫。

本文首發(fā)于我的博客：http://blog.didispace.com

系列導讀

• Spring Cloud構建微服務架構：服務注冊與發(fā)現(xiàn)（Eureka县好、Consul）
• Spring Cloud構建微服務架構：服務消費者（基礎）
• Spring Cloud構建微服務架構：服務消費者（Ribbon）
• Spring Cloud構建微服務架構：服務消費者（Feign）
• Spring Cloud構建微服務架構：分布式配置中心
• Spring Cloud構建微服務架構：服務容錯保護(Hystrix服務降級)
• Spring Cloud構建微服務架構：服務容錯保護(Hystrix依賴隔離)
• Spring Cloud構建微服務架構：服務容錯保護(Hystrix斷路器)
• Spring Cloud構建微服務架構：Hystrix監(jiān)控面板
• Spring Cloud構建微服務架構：Hystrix監(jiān)控數(shù)據(jù)聚合
• Spring Cloud構建微服務架構：服務網(wǎng)關（基礎）
• Spring Cloud構建微服務架構：服務網(wǎng)關（路由配置）
• Spring Cloud構建微服務架構：服務網(wǎng)關（過濾器）
• Spring Cloud構建微服務架構：消息驅動的微服務（入門）