### 訊息佇列 #### 基本概述 >MQ,Message Queue,基於TCP協議構建的**簡單協議**，區別於具體的**通訊協議**。 * 基於通訊協議定義和抽象的更高層次的通訊模型,一般都是**生產者和消費者模型**，又或者說**服務端和客戶端模型**。 * 生產者/消費者模型：一般通過定義生產者和消費者實現訊息通訊從而遮蔽複雜的底層通訊協議。應用於分散式應用系統，而且為之提供非同步解耦和削峰填谷的能力，同時也具備網際網路應用所需的海量堆積，高吞吐和可靠性重試機制的特性。 #### 核心概念 1. 訊息主題：Message Topic,一級訊息型別，生產者向其傳送訊息 2. 訊息生產者：Message Producer，訊息釋出者，或者訊息服務終端，負責生產訊息和傳送訊息到訊息主題 3. 訊息消費者：Message Consumer，訊息訂閱者，或者訊息客戶終端，負責從訊息主題接收並處理消費訊息 4. 訊息實體：Message Object,訊息物件，生產者向訊息主題傳送並最終傳送給訊息者的資料和屬性的符號以及組合 5. 訊息屬性：Message Attributes,訊息內容，生產者對訊息進行抽象和定義的相關屬性，包含Message Key 和Message Target 6. 訊息組：Message Group，訊息分類組別，通稱一類生產者和消費者，通常生產和消費同一類訊息，且訊息釋出和訂閱的邏輯基本一致 #### 程式設計思想 資料結構：訊息佇列的資料結構採用FIFO方式來定義與實現 設計模式：採用觀察者模式 >**觀察者模式**：定義物件間一對多的依賴關係，當一個物件的狀態發生改變時，所有依賴於它的物件都得到通知自動更新 #### 訊息流派： * 有Broker：基於訊息中介軟體，所有訊息通過中介軟體中轉，基於Broker把訊息推送給消費者[消費者主動輪詢] 1.重Topic：Kafka和RocketMq,Activemq，生產者把訊息傳送Key和資料到Broker，然後由Broker比較Key之後決定推送給具體的消費者[推送訊息給消費者] 2.輕Topic：Rabbitmq[AMQP],生產者把生產訊息且傳送Key和資料，消費者定義訂閱的佇列，Broker收到請求之後，按照一定的邏輯和規則計算出相應的Key對應的佇列，然後把訊息推送給具體的消費者[消費者向生產者訂閱訊息，消費者主動拉取訊息] * 無Broker：基於Socket網路[傳輸層]直連通訊，代表作ZeroMQ ### 基本概述  **NameServer**：**名稱伺服器**[MQ名稱空間伺服器],大致相當於 jndi技術，更新和發現 broker服務。用於儲存Broker相關元資訊，並給生產者和消費者查詢Broker訊息。每個Broker在啟動都會在名稱伺服器[NameServer]註冊,生產者在傳送訊息前會根據訊息主題到名稱伺服器查詢獲取Broker路由訊息，消費者也會定時獲取主題的路由訊息。 >ps：屬於無狀態服務設計，可橫向擴充套件，節點之間無通訊，可以部署多臺機器來標識偽叢集。 **Broker**:**訊息儲存中心**[訊息中轉角色]，負責儲存和轉發訊息。接收來自生產者的訊息並進行儲存，消費者從這拉取訊息。儲存與訊息相關的元資料，主要包括使用者組,訊息進度偏移量，佇列訊息等。其中Broker分為Master和Slave節點： 1. Master節點：可讀可寫 2. Slave節點：只可讀不可寫 其部署方式： * 單機Master:Broker重啟和宕機之後服務不可用,不建議生產使用 * 多機Master:所有訊息伺服器均是Master,沒有Slave。 優點：配置簡單，單個Master重啟和宕機維護對應用無影響 缺點：單機重啟和宕機期間，機器上未被消費的訊息在機器恢復期間不可訂閱，訊息實時性會受影響 * 多機Master多機Slave[同步雙寫]: 優點：資料和服務無單點故障，Master重啟和宕機訊息無延遲,資料和服務可用高 缺點：相對非同步複製效能較低，傳送訊息的延時較高 * 多機Master多機Slave[非同步複製]: 優點：訊息丟失較少，實時性高，Master重啟和宕機可繼續從Slave消費 缺點：Master重啟和宕機後在磁碟損壞會導致訊息丟失，但是情況比較少 >ps：一個Master可以對應多個Slave，但是一個Slave只能對應一個Master。Master和Slave的對應關係通過指定相同的BrokerName，不同的BrokerId來定義。BrokerId為0表示Master，BrokerId非0表示Slave。然後所有的Broker和Name Server上的節點建立長連線,定時註冊Topic資訊到所有Name Server。 **Producer**：**訊息生產者**->負責生產訊息,生產者向訊息伺服器傳送業務應用程式生成的訊息。主要有同步傳送和非同步傳送方式兩種，其中： * 同步傳送：訊息傳送方發出資料後，在訊息接收方傳送響應之後再發送下一個資料。一般適用於重要訊息通知場景[重要通知郵件，營銷簡訊推送] * 非同步傳送：訊息傳送方發出資料後，不用等待接收方發回響應，接著傳送下一個資料。一般適用於鏈路耗時較長，而對響應時間敏感的業務場景[視訊上傳通知 啟動自動轉碼服務處理通知] * 單向傳送：負責傳送訊息而不等待伺服器迴應且沒有回撥函式觸發。一般適用於對可靠性要求不高的業務場景[日誌收集] >ps：Producer與Name Server其中一個節點建立連線。定期從Name Server取Topic資訊。並與提供該Topic資訊的Master建立長連線。Producer也可以叢集部署。 **Consumer**：**訊息消費者** 負責消費訊息，從訊息伺服器拉取訊息並將其輸入使用者應用程式中。主要分為拉取型消費者和推送型消費者： 拉取型消費者：Pull Consumer->主動從訊息伺服器拉取訊息，只要批量拉取訊息,使用者就會啟動消費過程 推送型消費者：Push Consumer->封裝訊息的拉取,消費進度和其它內部維護工作，訊息到達之後便執行回撥介面留給使用者應用程式來實現。屬於被動消費型別，Push拉取時需要註冊訊息費者監聽器，當監聽器被觸發之後開始消費訊息。 >ps：Consumer 與Name Server 叢集中的其中一個節點（隨機選擇）建立長連線，定期從Name Server 取Topic 路由資訊，並向提供Topic服務的Master、Slave建立長連線，且定時向Master、Slave傳送心跳。Consumer既可以從Master訂閱訊息，也可以從Slave訂閱訊息，訂閱規則由Broker配置決定。 ### 部署Rocketmq 拉取rocketmq映象：docker pull foxiswho/rocketmq 1.查詢映象images：docker search rocketmq ``` [root@centos-meteor ~]# docker search rocketmq NAME DESCRIPTION STARS OFFICIAL AUTOMATED styletang/rocketmq-console-ng rocketmq-console-ng 20 rocketmqinc/rocketmq Image repository for Apache RocketMQ 17 foxiswho/rocketmq rocketmq 14 laoyumi/rocketmq 10 [OK] xlxwhy/rocketmq alibaba's rocketmq 4 huanwei/rocketmq-broker 2 2019liurui/rocketmq-broker RocketMQ broker image for RocketMQ-Operator 1 2019liurui/rocketmq-namesrv RocketMQ name service image for RocketMQ-Ope… 1 apacherocketmq/rocketmq Docker Image for Apache RocketMQ 1 rocketmqinc/rocketmq-operator The Kubernetes operator for RocketMQ 0 2019liurui/rocketmq-operator Kubernetes Operator for RocketMQ ! 0 apacherocketmq/rocketmq-operator RocketMQ Operator is to manage RocketMQ serv… 0 coder4/rocketmq rocketmq 0 [OK] rocketmqinc/rocketmq-namesrv Customized RocketMQ Name Server Image for Ro… 0 rocketmqinc/rocketmq-broker Customized RocketMQ Broker Image for RocketM… 0 slpcat/rocketmq-console-ng 0 huanwei/rocketmq 0 huanwei/rocketmq-broker-k8s 0 king019/rocketmq rocketmq 0 pengzu/rocketmq-console-ng web console for rocketmq ,this code is from … 0 fengzt/rocketmq-broker apache rocketmq 4.2.0 broker server(官方文件… 0 huanwei/rocketmq-operator 0 slpcat/rocketmq 0 fengzt/rocketmq-nameserver apache rocketmq 4.2.0 nameserver 0 icyblazek/rocketmq RocketMQ 0 [root@centos-meteor ~]# ``` 2.執行：docker pull foxiswho/rocketmq 3.建立docker儲存根目錄並且授權: mkdir docker && chmod -R 777 docker/ ``` [root@centos-meteor /]# cd docker/ [root@centos-meteor docker]# pwd /docker [root@centos-meteor docker]# ``` 4.部署名稱伺服器rocketmq-namesrv-server[9876]: #rocketmq-namesrv-server docker run -itd --restart=always --privileged=true -p 9876:9876 --name rocketmq-namesrv-server -v /docker/rocketmq/namesrv/logs:/root/rocketmq/logs -v /docker/rocketmq/namesrv/store:/root/rocketmq/store -e "MAX_POSSIBLE_HEAP=100000000" -e "JAVA_OPTS=-Duser.home=/opt" -e "JAVA_OPT_EXT=-server -Xms512m -Xmx512m -Xmn256m" rocketmqinc/rocketmq sh mqnamesrv ``` docker run -itd --restart=always --privileged=true -p 9876:9876 --name rocketmq-namesrv-server -v /docker/rocketmq/namesrv/logs:/root/rocketmq/logs -v /docker/rocketmq/namesrv/store:/root/rocketmq/store -e "MAX_POSSIBLE_HEAP=100000000" -e "JAVA_OPTS=-Duser.home=/opt" -e "JAVA_OPT_EXT=-server -Xms512m -Xmx512m -Xmn256m" rocketmqinc/rocketmq sh mqnamesrv ``` >ps：建立/docker/rocketmq/namesrv/logs和/docker/rocketmq/namesrv/store目錄 5.部署訊息伺服器rocketmq-broker-server[10911]： #rocketmq-broker-server docker run -itd --restart=always --privileged=true -p 10909:10909 -p 10911:10911 -p 10912:10912 --name rocketmq-broker-server --link rocketmq-namesrv-server:namesrv -v /docker/rocketmq/broker/logs:/root/rocketmq/logs -v /docker/rocketmq/broker/store:/root/rocketmq/store -v /docker/rocketmq/broker/conf/broker.conf:/opt/rocketmq-4.4.0/conf/broker.conf -e "NAMESRV_ADDR=namesrv:9876" -e "MAX_POSSIBLE_HEAP=200000000" -e "JAVA_OPTS=-Duser.home=/opt" -e "JAVA_OPT_EXT=-server -Xms512m -Xmx512m -Xmn256m" rocketmqinc/rocketmq sh mqbroker -c /opt/rocketmq-4.4.0/conf/broker.conf ``` docker run -itd --restart=always --privileged=true -p 10909:10909 -p 10911:10911 -p 10912:10912 --name rocketmq-broker-server --link rocketmq-namesrv-server:namesrv -v /docker/rocketmq/broker/logs:/root/rocketmq/logs -v /docker/rocketmq/broker/store:/root/rocketmq/store -v /docker/rocketmq/broker/conf/broker.conf:/opt/rocketmq-4.4.0/conf/broker.conf -e "NAMESRV_ADDR=namesrv:9876" -e "MAX_POSSIBLE_HEAP=200000000" -e "JAVA_OPTS=-Duser.home=/opt" -e "JAVA_OPT_EXT=-server -Xms512m -Xmx512m -Xmn256m" rocketmqinc/rocketmq sh mqbroker -c /opt/rocketmq-4.4.0/conf/broker.conf ``` >ps： >1.建立/docker/rocketmq/broker/logs和/docker/rocketmq/broker/store目錄 >2.編寫broker.conf配置檔案： ``` brokerClusterName = rocketmq-cluster brokerName = broker-server brokerId = 0 deleteWhen = 04 fileReservedTime = 48 # Broker 的角色 # - ASYNC_MASTER 非同步複製Master # - SYNC_MASTER 同步雙寫Master # - SLAVE brokerRote=ASYNC_MASTER brokerRole = ASYNC_MASTER flushDiskType = ASYNC_FLUSH # 如果是本地程式呼叫雲主機 mq，這個需要設定成 雲主機 IP brokerIP1=Server-IP #限制的訊息大小 maxMessageSize=65536 # 檢測物理檔案磁碟空間 diskMaxUsedSpaceRatio=88 mapedFileSizeCommitLog=1073741824 mapedFileSizeConsumeQueue=50000000 #併發send執行緒數，多執行緒來發送訊息可能會出現broker busy sendMessageThreadPoolNums=128 useReentrantLockWhenPutMessage=true #在傳送訊息時，自動建立伺服器不存在的topic，預設建立的佇列數 defaultTopicQueueNums=8 highSpeedMode=false commercialBaseCount=1 maxErrorRateOfBloomFilter=20 accessMessageInMemoryMaxRatio=40 #無讀寫客戶端存活時間 clientChannelMaxIdleTimeSeconds=120 flushDelayOffsetInterval=10000 serverSocketRcvBufSize=131072 #單次 Pull 訊息(記憶體)傳輸的 最大位元組數 maxTransferBytesOnMessageInMemory=262144 clientManageThreadPoolNums=32 serverChannelMaxIdleTimeSeconds=120 serverCallbackExecutorThreads=0 enablePropertyFilter=false transientStorePoolSize=5 enableConsumeQueueExt=false #rocketmq server config serverPooledByteBufAllocatorEnable=true serverSocketRcvBufSize=131072 #rocketmq client config ``` 6.部署控制後臺rocketmq-consloe-server[8082]: #rocketmq-consloe-server docker run -itd -p 8082:8080 --restart=always --privileged=true --name rocketmq-console-server -v /docker/rocketmq/console/data:/tmp -e "JAVA_OPTS=-Drocketmq.namesrv.addr=47.104.22.10:9876 -Dcom.rocketmq.sendMessageWithVIPChannel=false" -e "JAVA_OPT_EXT=-server -Xms512m -Xmx512m -Xmn256m -Duser.home=/opt" styletang/rocketmq-console-ng:latest ``` docker run -itd -p 8082:8080 --restart=always --privileged=true --name rocketmq-console-server -v /docker/rocketmq/console/data:/tmp -e "JAVA_OPTS=-Drocketmq.namesrv.addr=Server-IP:9876 -Dcom.rocketmq.sendMessageWithVIPChannel=false" -e "JAVA_OPT_EXT=-server -Xms512m -Xmx512m -Xmn256m -Duser.home=/opt" styletang/rocketmq-console-ng:latest ``` >ps： >1.建立/docker/rocketmq/console/data目錄 >2.Server-IP為broker.conf配置brokerIP1值 7.最終部署結果：  ### 整合Rocketmq開發實戰 1.配置Rocketmq的Maven依賴： ```xml ``` >ps： >1.目前rocketmq最新版本是4.7.1： ```xml ``` >2.這裡選擇的是rocketmq版本[4.4.0],因為部署是4.4.0版本，推薦部署版本相同的rocketmq 2.建立pivotal-cloud-queue工程:  >ps：建立訊息佇列統一模組工程，在業務應用程式服務模組依賴該工程，並封裝訊息生產者[Producer]和訊息消費者[Consumer]處理類。 3.封裝rocketmq屬性配置類: 新增spring-boot-configuration-processor依賴: ```xml ``` 編寫RocketmqProperties屬性配置類:  ```java package com.pivotal.cloud.queue.properties; import lombok.Data; import org.springframework.boot.context.properties.ConfigurationProperties; import org.springframework.context.annotation.Configuration; /** * @className: com.pivotal.cloud.queue.properties.RocketmqProperties * @title: RocketmqProperties * @description: 封裝Pivotal專案RocketmqProperties類 * @content: PivotalCloud專案系統RocketmqProperties自定義屬性配置類 * @author: marklin * @datetime: 2020-07-07 01:32 * @version: 1.0.0 * @copyright: Copyright © 2018-2020 Pivotal Systems Incorporated. All rights reserved. */ @Data @Configuration @ConfigurationProperties(prefix = "pivotal.cloud.rocketmq") public class RocketmqProperties { /** * rocketmq訊息佇列生產者-Producer */ private final Producer producer = new Producer(); /** * rocketmq訊息佇列消費者-Consumer */ private final Consumer consumer = new Consumer(); @Data public static class Producer { } @Data public static class Consumer { } } ``` 編寫RocketmqConfiguration配置類: [](https://imgchr.com/i/UFM2Os) ```java package com.pivotal.cloud.queue.configuration; import com.pivotal.cloud.queue.properties.RocketmqProperties; import lombok.AllArgsConstructor; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.context.annotation.Configuration; /** * @className: com.pivotal.cloud.queue.configuration.RocketmqConfiguration * @title: RocketmqConfiguration * @description: 封裝Pivotal專案RocketmqConfiguration類 * @content: //TODO * @author: marklin * @datetime: 2020-07-07 02:25 * @version: 1.0.0 * @copyright: Copyright © 2018-2020 Pivotal Systems Incorporated. All rights reserved. */ @AllArgsConstructor @Configuration(proxyBeanMethods = false) @EnableConfigurationProperties({RocketmqProperties.class}) public class RocketmqConfiguration { } ``` 編寫RocketmqTemplate生產者模板類: ```java package com.pivotal.cloud.queue.template; /** * @className: com.pivotal.cloud.queue.template.RocketmqTemplate * @title: RocketmqTemplate * @description: 封裝Pivotal專案RocketmqTemplate類 * @content: PivotalCloud專案系統RocketmqTemplate生產者模板類 * @author: marklin * @datetime: 2020-07-07 02:48 * @version: 1.0.0 * @copyright: Copyright © 2018-2020 Pivotal Systems Incorporated. All rights reserved. */ public class RocketmqTemplate { } ``` 編寫RocketmqListener消費者監聽器類:  ```java package com.pivotal.cloud.queue.listener; /** * @className: com.pivotal.cloud.queue.listener.RocketmqListener * @title: RocketmqListener * @description: 封裝Pivotal專案RocketmqListener類 * @content: PivotalCloud專案系統RocketmqListener消費者監聽器類 * @author: marklin * @datetime: 2020-07-07 02:52 * @version: 1.0.0 * @copyright: Copyright © 2018-2020 Pivotal Systems Incorporated. All rights reserved. */ public interface RocketmqListener { } ``` 編寫META-INF/spring.factories工廠類: [](https://imgchr.com/i/UFM5kV) ```factories org.springframework.boot.autoconfigure.EnableAutoConfiguration=\ com.pivotal.cloud.queue.configuration.RocketmqConfiguration,\ com.pivotal.cloud.queue.configuration.RabbitmqConfiguration,\ com.pivotal.cloud.queue.configuration.ActivemqConfiguration,\ com.pivotal.cloud.queue.configuration.KafkaConfiguration, ``` 版權宣告：本文為博主原創文章，遵循相關版權協議，如若轉載或者分享請附上原文出處連結和連結