如何在優雅地Spring 中實現消息的發送和消費

本文將對rocktmq-spring-boot的設計實現作一個簡單的介紹，讀者能夠經過本文了解將RocketMQ Client端集成爲spring-boot-starter框架的開發細節，而後經過一個簡單的示例來一步一步的講解如何使用這個spring-boot-starter工具包來配置，發送和消費RocketMQ消息。

做者簡介：遼天，阿里巴巴技術專家，Apache RocketMQ 內核控，擁有多年分佈式系統研發經驗，對Microservice、Messaging和Storage等領域有深入理解， 目前專一 RocketMQ 內核優化以及 Messaging 生態建設。

經過本文，您將瞭解到：

• Spring的消息框架介紹
• rocketmq-spring-boot具體實現
• 使用示例

插播一條廣告：本週六下午，Apache RocketMQ 開發者沙龍未來到杭州，歡迎你們到現場，活動詳情請點擊「閱讀原文」。

前言

上世紀90年代末，隨着Java EE(Enterprise Edition)的出現，特別是Enterprise Java Beans的使用須要複雜的描述符配置和死板複雜的代碼實現，增長了廣大開發者的學習曲線和開發成本，由此基於簡單的XML配置和普通Java對象(Plain Old Java Objects)的Spring技術應運而生，依賴注入(Dependency Injection), 控制反轉(Inversion of Control)和麪向切面編程(AOP)的技術更加敏捷地解決了傳統Java企業及版本的不足。

隨着Spring的持續演進，基於註解(Annotation)的配置逐漸取代了XML文件配置， 2014年4月1日，Spring Boot 1.0.0正式發佈，它基於「約定大於配置」（Convention over configuration)這一理念來快速地開發、測試、運行和部署Spring應用，並能經過簡單地與各類啓動器(如 spring-boot-web-starter)結合，讓應用直接以命令行的方式運行，不需再部署到獨立容器中。這種簡便直接快速構建和開發應用的過程，可使用約定的配置而且簡化部署，受到愈來愈多的開發者的歡迎。

Apache RocketMQ是業界知名的分佈式消息和流處理中間件，簡單地理解，它由Broker服務器和客戶端兩部分組成：

其中客戶端一個是消息發佈者客戶端(Producer)，它負責向Broker服務器發送消息；
另一個是消息的消費者客戶端(Consumer)，多個消費者能夠組成一個消費組，來訂閱和拉取消費Broker服務器上存儲的消息。

爲了利用Spring Boot的快速開發和讓用戶可以更靈活地使用RocketMQ消息客戶端，Apache RocketMQ社區推出了spring-boot-starter實現。隨着分佈式事務消息功能在RocketMQ 4.3.0版本的發佈，近期升級了相關的spring-boot代碼，經過註解方式支持分佈式事務的回查和事務消息的發送。

本文將對當前的設計實現作一個簡單的介紹，讀者能夠經過本文了解將RocketMQ Client端集成爲spring-boot-starter框架的開發細節，而後經過一個簡單的示例來一步一步的講解如何使用這個spring-boot-starter工具包來配置，發送和消費RocketMQ消息。

Spring 中的消息框架

順便在這裏討論一下在Spring中關於消息的兩個主要的框架，即Spring Messaging和Spring Cloud Stream。它們都可以與Spring Boot整合並提供了一些參考的實現。和全部的實現框架同樣，消息框架的目的是實現輕量級的消息驅動的微服務，能夠有效地簡化開發人員對消息中間件的使用複雜度，讓系統開發人員能夠有更多的精力關注於核心業務邏輯的處理。

2.1 Spring Messaging

Spring Messaging是Spring Framework 4中添加的模塊，是Spring與消息系統集成的一個擴展性的支持。它實現了從基於JmsTemplate的簡單的使用JMS接口到異步接收消息的一整套完整的基礎架構，Spring AMQP提供了該協議所要求的相似的功能集。 在與Spring Boot的集成後，它擁有了自動配置能力，可以在測試和運行時與相應的消息傳遞系統進行集成。

單純對於客戶端而言，Spring Messaging提供了一套抽象的API或者說是約定的標準，對消息發送端和消息接收端的模式進行規定，不一樣的消息中間件提供商能夠在這個模式下提供本身的Spring實現：在消息發送端須要實現的是一個XXXTemplate形式的Java Bean，結合Spring Boot的自動化配置選項提供多個不一樣的發送消息方法；在消息的消費端是一個XXXMessageListener接口（實現方式一般會使用一個註解來聲明一個消息驅動的POJO），提供回調方法來監聽和消費消息，這個接口一樣可使用Spring Boot的自動化選項和一些定製化的屬性。

若是有興趣深刻的瞭解Spring Messaging及針對不一樣的消息產品的使用，推薦閱讀這個文件。參考Spring Messaging的既有實現，RocketMQ的spring-boot-starter中遵循了相關的設計模式並結合RocketMQ自身的功能特色提供了相應的API(如，順序，異步和事務半消息等)。

2.2 Spring Cloud Stream

Spring Cloud Stream結合了Spring Integration的註解和功能，它的應用模型以下：

該圖片引自spring cloud stream

Spring Cloud Stream框架中提供一個獨立的應用內核，它經過輸入(@Input)和輸出(@Output)通道與外部世界進行通訊，消息源端(Source)經過輸入通道發送消息，消費目標端(Sink)經過監聽輸出通道來獲取消費的消息。這些通道經過專用的Binder實現與外部代理鏈接。開發人員的代碼只須要針對應用內核提供的固定的接口和註解方式進行編程，而不須要關心運行時具體的Binder綁定的消息中間件。在運行時，Spring Cloud Stream可以自動探測並使用在classpath下找到的Binder。

這樣開發人員能夠輕鬆地在相同的代碼中使用不一樣類型的中間件：僅僅須要在構建時包含進不一樣的Binder。在更加複雜的使用場景中，也能夠在應用中打包多個Binder並讓它本身選擇Binder，甚至在運行時爲不一樣的通道使用不一樣的Binder。

Binder抽象使得Spring Cloud Stream應用能夠靈活的鏈接到中間件，加之Spring Cloud Stream使用利用了Spring Boot的靈活配置配置能力，這樣的配置能夠經過外部配置的屬性和Spring Boo支持的任何形式來提供（包括應用啓動參數、環境變量和application.yml或者application.properties文件），部署人員能夠在運行時動態選擇通道鏈接destination（例如，Kafka的topic或者RabbitMQ的exchange）。

Binder SPI的方式來讓消息中間件產品使用可擴展的API來編寫相應的Binder，並集成到Spring Cloud Steam環境，目前RocketMQ尚未提供相關的Binder，咱們計劃在下一步將完善這一功能，也但願社區裏有這方面經驗的同窗積極嘗試，貢獻PR或建議。

spring-boot-starter的實現

在開始的時候咱們已經知道，spring boot starter構造的啓動器對於使用者是很是方便的，使用者只要在pom.xml引入starter的依賴定義，相應的編譯，運行和部署功能就所有自動引入。所以經常使用的開源組件都會爲Spring的用戶提供一個spring-boot-starter封裝給開發者，讓開發者很是方便集成和使用，這裏咱們詳細的介紹一下RocketMQ（客戶端）的starter實現過程。

3.1. spring-boot-starter的實現步驟

對於一個spring-boot-starter實現須要包含以下幾個部分：

1. 在pom.xml的定義

• 定義最終要生成的starter組件信息

<groupId>org.apache.rocketmq</groupId>
<artifactId>spring-boot-starter-rocketmq</artifactId>
<version>1.0.0-SNAPSHOT</version>
複製代碼

• 定義依賴包，

它分爲兩個部分: A、Spring自身的依賴包； B、RocketMQ的依賴包

<dependencies>
    <!-- spring-boot-start internal depdencies -->
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter</artifactId>
    </dependency>         
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
    
    
    <!-- rocketmq dependencies -->
    <dependency>
        <groupId>org.apache.rocketmq</groupId>
        <artifactId>rocketmq-client</artifactId>
        <version>${rocketmq-version}</version>
    </dependency>
</dependencies>    
    <dependencyManagement>
    <dependencies>
        <!-- spring-boot-start parent depdency definition --> 
        <dependency>
            <groupId>org.springframework.boot</groupId>
            <artifactId>spring-boot-starter-parent</artifactId>
            <version>${spring.boot.version}</version>
            <type>pom</type>
            <scope>import</scope>
        </dependency>
    </dependencies>
</dependencyManagement>
複製代碼

1. 配置文件類

定義應用屬性配置文件類RocketMQProperties,這個Bean定義一組默認的屬性值。用戶在使用最終的starter時，能夠根據這個類定義的屬性來修改取值，固然不是直接修改這個類的配置，而是spring-boot應用中對應的配置文件：src/main/resources/application.properties.

1. 定義自動加載類

定義 src/resources/META-INF/spring.factories文件中的自動加載類， 其目的是讓spring boot更具文中中所指定的自動化配置類來自動初始化相關的Bean，Component或Service，它的內容以下：

org.springframework.boot.autoconfigure.EnableAutoConfiguration=\
org.apache.rocketmq.spring.starter.RocketMQAutoConfiguration
複製代碼

在RocketMQAutoConfiguration類的具體實現中，定義開放給用戶直接使用的Bean對象. 包括：

• RocketMQProperties 加載應用屬性配置文件的處理類；
• RocketMQTemplate 發送端用戶發送消息的發送模板類；
• ListenerContainerConfiguration 容器Bean負責發現和註冊消費端消費實現接口類，這個類要求：由@RocketMQMessageListener註解標註；實現RocketMQListener泛化接口。

1. 最後具體的RocketMQ相關的封裝
   在發送端（producer）和消費端(consumer)客戶端分別進行封裝，在當前的實現版本提供了對Spring Messaging接口的兼容方式。

3.2. 消息發送端實現

1. 普通發送端

發送端的代碼封裝在RocketMQTemplate POJO中，下圖是發送端的相關代碼的調用關係圖：

爲了與Spring Messaging的發送模板兼容，在RocketMQTemplate集成了AbstractMessageSendingTemplate抽象類，來支持相關的消息轉換和發送方法，這些方法最終會代理給doSend()方法；doSend()以及RocoketMQ所特有的一些方法如異步，單向和順序等方法直接添加到RoketMQTempalte中，這些方法直接代理調用到RocketMQ的Producer API來進行消息的發送。

1. 事務消息發送端

對於事務消息的處理，在消息發送端進行了部分的擴展，參考下圖的調用關係類圖：

RocketMQTemplate里加入了一個發送事務消息的方法sendMessageInTransaction(), 而且最終這個方法會代理到RocketMQ的TransactionProducer進行調用，在這個Producer上會註冊其關聯的TransactionListener實現類，以便在發送消息後可以對TransactionListener裏的方法實現進行調用。

3.3. 消息消費端實現

在消費端Spring-Boot應用啓動後，會掃描全部包含@RocketMQMessageListener註解的類(這些類須要集成RocketMQListener接口，並實現onMessage()方法)，這個Listener會一對一的被放置到DefaultRocketMQListenerContainer容器對象中，容器對象會根據消費的方式(併發或順序)，將RocketMQListener封裝到具體的RocketMQ內部的併發或者順序接口實現。在容器中建立RocketMQ Consumer對象，啓動並監聽定製的Topic消息，若是有消費消息，則回調到Listener的onMessage()方法。

使用示例

上面的一章介紹了RocketMQ在spring-boot-starter方式的實現，這裏經過一個最簡單的消息發送和消費的例子來介紹如何使這個rocketmq-spring-boot-starter。

4.1 RocketMQ服務端的準備

1. 啓動NameServer和Broker

要驗證RocketMQ的Spring-Boot客戶端，首先要確保RocketMQ服務正確的下載並啓動。能夠參考RocketMQ主站的快速開始來進行操做。確保啓動NameServer和Broker已經正確啓動。

1. 建立實例中所須要的Topics

在執行啓動命令的目錄下執行下面的命令行操做

bash bin/mqadmin updateTopic -c DefaultCluster -t string-topic
複製代碼

4.2. 編譯rocketmq-spring-boot-starter

目前的spring-boot-starter依賴尚未提交的Maven的中心庫，用戶使用前須要自行下載git源碼，而後執行mvn clean install 安裝到本地倉庫。

git clone https://github.com/apache/rocketmq-externals.git
cd rocketmq-spring-boot-starter
mvn clean install
複製代碼

4.3. 編寫客戶端代碼

用戶若是使用它，須要在消息的發佈和消費客戶端的maven配置文件pom.xml中添加以下的依賴：

<properties>   <spring-boot-starter-rocketmq-version>1.0.0-SNAPSHOT</spring-boot-starter-rocketmq-version>
</properties>

<dependency>
   <groupId>org.apache.rocketmq</groupId>
   <artifactId>spring-boot-starter-rocketmq</artifactId>
   <version>${spring-boot-starter-rocketmq-version}</version>
</dependency>
複製代碼

屬性spring-boot-starter-rocketmq-version的取值爲：1.0.0-SNAPSHOT， 這與上一步驟中執行安裝到本地倉庫的版本一致。

1. 消息發送端的代碼

發送端的配置文件application.properties

# 定義name-server地址
spring.rocketmq.name-server=localhost:9876
# 定義發佈者組名
spring.rocketmq.producer.group=my-group1
# 定義要發送的topic
spring.rocketmq.topic=string-topic
複製代碼

發送端的Java代碼

import org.apache.rocketmq.spring.starter.core.RocketMQTemplate;
...

@SpringBootApplication
public class ProducerApplication implements CommandLineRunner {
    // 聲明並引用RocketMQTemplate
    @Resource
    private RocketMQTemplate rocketMQTemplate;

    // 使用application.properties裏定義的topic屬性
    @Value("${spring.rocketmq.springTopic}")
    private String springTopic;
    
    public static void main(String[] args){
        SpringApplication.run(ProducerApplication.class, args);
    }
    
    public void run(String... args) throws Exception {
        // 以同步的方式發送字符串消息給指定的topic
        SendResult sendResult = rocketMQTemplate.syncSend(springTopic, "Hello, World!");
        // 打印發送結果信息
        System.out.printf("string-topic syncSend1 sendResult=%s %n", sendResult);
    }
}
複製代碼

1. 消息消費端代碼

消費端的配置文件application.properties

# 定義name-server地址
spring.rocketmq.name-server=localhost:9876
# 定義發佈者組名
spring.rocketmq.consumer.group=my-customer-group1
# 定義要發送的topic
spring.rocketmq.topic=string-topic
複製代碼

消費端的Java代碼

@SpringBootApplication
public class ConsumerApplication {

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


// 聲明消費消息的類，並在註解中指定，相關的消費信息
@Service
@RocketMQMessageListener(topic = "${spring.rocketmq.topic}", consumerGroup = "${spring.rocketmq.consumer.group}")
class StringConsumer implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        System.out.printf("------- StringConsumer received: %s %f", message);
    }
}
複製代碼

原文連接

本文爲雲棲社區原創內容，未經容許不得轉載。