[[391523]]

環(huán)境：springboot2.3.9 + RocketMQ4.8.0

RocketMQ機(jī)構(gòu)及概念

1 消息模型(Message Model)

RocketMQ主要由 Producer、Broker、Consumer 三部分組成，其中Producer 負(fù)責(zé)生產(chǎn)消息，Consumer 負(fù)責(zé)消費(fèi)消息，Broker 負(fù)責(zé)存儲消息。Broker 在實際部署過程中對應(yīng)一臺服務(wù)器，每個 Broker 可以存儲多個Topic的消息，每個Topic的消息也可以分片存儲于不同的 Broker。Message Queue 用于存儲消息的物理地址，每個Topic中的消息地址存儲于多個 Message Queue 中。ConsumerGroup 由多個Consumer 實例構(gòu)成。

2 消息生產(chǎn)者(Producer)

負(fù)責(zé)生產(chǎn)消息，一般由業(yè)務(wù)系統(tǒng)負(fù)責(zé)生產(chǎn)消息。一個消息生產(chǎn)者會把業(yè)務(wù)應(yīng)用系統(tǒng)里產(chǎn)生的消息發(fā)送到broker服務(wù)器。RocketMQ提供多種發(fā)送方式，同步發(fā)送、異步發(fā)送、順序發(fā)送、單向發(fā)送。同步和異步方式均需要Broker返回確認(rèn)信息，單向發(fā)送不需要。

3 消息消費(fèi)者(Consumer)

負(fù)責(zé)消費(fèi)消息，一般是后臺系統(tǒng)負(fù)責(zé)異步消費(fèi)。一個消息消費(fèi)者會從Broker服務(wù)器拉取消息、并將其提供給應(yīng)用程序。從用戶應(yīng)用的角度而言提供了兩種消費(fèi)形式：拉取式消費(fèi)、推動式消費(fèi)。

4 主題(Topic)

表示一類消息的集合，每個主題包含若干條消息，每條消息只能屬于一個主題，是RocketMQ進(jìn)行消息訂閱的基本單位。

5 代理服務(wù)器(Broker Server)

消息中轉(zhuǎn)角色，負(fù)責(zé)存儲消息、轉(zhuǎn)發(fā)消息。代理服務(wù)器在RocketMQ系統(tǒng)中負(fù)責(zé)接收從生產(chǎn)者發(fā)送來的消息并存儲、同時為消費(fèi)者的拉取請求作準(zhǔn)備。代理服務(wù)器也存儲消息相關(guān)的元數(shù)據(jù)，包括消費(fèi)者組、消費(fèi)進(jìn)度偏移和主題和隊列消息等。如下圖：

6 名字服務(wù)(Name Server)

名稱服務(wù)充當(dāng)路由消息的提供者。生產(chǎn)者或消費(fèi)者能夠通過名字服務(wù)查找各主題相應(yīng)的Broker IP列表。多個Namesrv實例組成集群，但相互獨(dú)立，沒有信息交換。

7 拉取式消費(fèi)(Pull Consumer)

Consumer消費(fèi)的一種類型，應(yīng)用通常主動調(diào)用Consumer的拉消息方法從Broker服務(wù)器拉消息、主動權(quán)由應(yīng)用控制。一旦獲取了批量消息，應(yīng)用就會啟動消費(fèi)過程。

8 推動式消費(fèi)(Push Consumer)

Consumer消費(fèi)的一種類型，該模式下Broker收到數(shù)據(jù)后會主動推送給消費(fèi)端，該消費(fèi)模式一般實時性較高。

9 生產(chǎn)者組(Producer Group)

同一類Producer的集合，這類Producer發(fā)送同一類消息且發(fā)送邏輯一致。如果發(fā)送的是事務(wù)消息且原始生產(chǎn)者在發(fā)送之后崩潰，則Broker服務(wù)器會聯(lián)系同一生產(chǎn)者組的其他生產(chǎn)者實例以提交或回溯消費(fèi)。

10 消費(fèi)者組(Consumer Group)

同一類Consumer的集合，這類Consumer通常消費(fèi)同一類消息且消費(fèi)邏輯一致。消費(fèi)者組使得在消息消費(fèi)方面，實現(xiàn)負(fù)載均衡和容錯的目標(biāo)變得非常容易。要注意的是，消費(fèi)者組的消費(fèi)者實例必須訂閱完全相同的Topic。RocketMQ 支持兩種消息模式：集群消費(fèi)(Clustering)和廣播消費(fèi)(Broadcasting)。

11 集群消費(fèi)(Clustering)

集群消費(fèi)模式下,相同Consumer Group的每個Consumer實例平均分?jǐn)傁ⅰ?/p>

12 廣播消費(fèi)(Broadcasting)

廣播消費(fèi)模式下，相同Consumer Group的每個Consumer實例都接收全量的消息。

13 普通順序消息(Normal Ordered Message)

普通順序消費(fèi)模式下，消費(fèi)者通過同一個消費(fèi)隊列收到的消息是有順序的，不同消息隊列收到的消息則可能是無順序的。

14 嚴(yán)格順序消息(Strictly Ordered Message)

嚴(yán)格順序消息模式下，消費(fèi)者收到的所有消息均是有順序的。

15 消息(Message)

消息系統(tǒng)所傳輸信息的物理載體，生產(chǎn)和消費(fèi)數(shù)據(jù)的最小單位，每條消息必須屬于一個主題。RocketMQ中每個消息擁有唯一的Message ID，且可以攜帶具有業(yè)務(wù)標(biāo)識的Key。系統(tǒng)提供了通過Message ID和Key查詢消息的功能。

16 標(biāo)簽(Tag)

為消息設(shè)置的標(biāo)志，用于同一主題下區(qū)分不同類型的消息。來自同一業(yè)務(wù)單元的消息，可以根據(jù)不同業(yè)務(wù)目的在同一主題下設(shè)置不同標(biāo)簽。標(biāo)簽?zāi)軌蛴行У乇３执a的清晰度和連貫性，并優(yōu)化RocketMQ提供的查詢系統(tǒng)。消費(fèi)者可以根據(jù)Tag實現(xiàn)對不同子主題的不同消費(fèi)邏輯，實現(xiàn)更好的擴(kuò)展性。

ActiveMQ，Kafka，RocketMQ對比：

RocketMQ服務(wù)

1 下載RocketMQ

2 配置環(huán)境變量

3 啟動Name Server

4 啟動 Broker

5 通過命令行發(fā)送 & 接收消息

設(shè)置環(huán)境變量：

C:\Users\MSI-NB>set NAMESRV_ADDR=localhost:9876 

發(fā)送消息：

C:\Users\MSI-NB>tools org.apache.rocketmq.example.quickstart.Producer 

接收消息：

SpringBoot整合RocketMQ入門

依賴：

<dependency> 
  <groupId>org.apache.rocketmq</groupId> 
  <artifactId>rocketmq-spring-boot-starter</artifactId> 
  <version>2.2.0</version> 
</dependency> 

配置文件：

rocketmq: 
  nameServer: localhost:9876 
  producer: 
    group: demo-mq 

生產(chǎn)者：

@Service 
public class ProducerService { 
     
    @Resource 
    private RocketMQTemplate rocketMQTemplate ; 
     
    public void send(String message) { 
        rocketMQTemplate.convertAndSend("test-topic", message); 
    } 
     
} 

消費(fèi)者：

@RocketMQMessageListener(topic = "test-topic", consumerGroup = "consumer01-group") 
@Component 
public class ConsumerListener implements RocketMQListener<String> { 
 
    @Override 
    public void onMessage(String message) { 
        System.out.println("接收到消息：" + message) ; 
    } 
 
} 

這里的topic要和發(fā)送端設(shè)置的一致，consumerGroup可隨意。

發(fā)送接口:

@RestController 
@RequestMapping("/messages") 
public class MessageController { 
     
    @Resource 
    private ProducerService ps ; 
     
    @GetMapping("") 
    public Object send(String message) { 
        ps.send(message) ; 
        return "send success" ; 
    } 
     
} 

測試：

發(fā)送消息指定tags

發(fā)送時：

rocketMQTemplate.convertAndSend("test-topic:tag1", message); 

接收時：

@RocketMQMessageListener(topic = "test-topic", consumerGroup = "consumer01-group", selectorExpression = "tag1") 

selectorExpression：默認(rèn)是 “*” ，這里指定與發(fā)送的一致;

這里看下源碼：

RocketMQUtil.java

這里topic與tags是用冒號 ":" 分割的，tags就是取的數(shù)組的第二個。