RocketMQ基礎

  • 2019 年 12 月 31 日
  • 筆記

「 RockeMQ是阿里巴巴自主研發的消息中間件，實現業務削峰，分散式事務等，已捐獻給Apache，成為MQ家族的重要成員。」

01

—

安裝RocketMQ

RocketMQ集群的部署可以分為很多種模式：

nameserver可以多個；

broker也可以部署多個：兩主兩從、兩主無從等，本文採用兩主無從的模式；

# 下載  wget http://mirrors.tuna.tsinghua.edu.cn/apache/rocketmq/4.6.0/rocketmq-all-4.6.0-bin-release.zip  # 解壓  unzip rocketmq-all-4.6.0-bin-release.zip  # 重命名  mv rocketmq-all-4.6.0-bin-release/ rocketmq
# 配置環境變數，三個節點都需要配置！！！  vim .bash_profile  export ROCKETMQ=/opt/install/rocketmq/  export PATH=$PATH:$ROCKETMQ/bin  source .bash_profile  # 修改日誌文件位置  sed 's#${user.home}#/opt/install/rocketmq#g' *.xml -i  # 分發到其他節點  scp -P 12324 -r rocketmq node02:$PWD  scp -P 12324 -r rocketmq node03:$PWD    在node01上啟動nameserver  nohup sh /opt/install/rocketmq/bin/mqnamesrv >/dev/null 2>&1 &
# broker配置  vim runbroker.sh  # 默認 ${JAVA_OPT} -server -Xms 8g -Xmx2 8g -Xmn 4g  # 可能機器記憶體不足，導致啟動不起來，根據機器配置調小點  JAVA_OPT="${JAVA_OPT} -server -Xms2g -Xmx2g -Xmn1g"  # 修改配置文件，這裡修改conf/2m-noslave/.properties  # 註：另一個節點是node03  brokerName=node02  # 0 （master） 非0 slave  brokerId=0  brokerRole=ASYNC_MASTER  #所屬集群名字  brokerClusterName=rocketmq-cluster  #nameServer地址，分號分割  namesrvAddr=10.255.20.147:9876  #在發送消息時，自動創建伺服器不存在的topic，默認創建的隊列數  defaultTopicQueueNums=4  #是否允許 Broker 自動創建Topic，建議線下開啟，線上關閉  autoCreateTopicEnable=true  #是否允許 Broker 自動創建訂閱組，建議線下開啟，線上關閉  autoCreateSubscriptionGroup=true  #Broker 對外服務的監聽埠  listenPort=10911  #刪除文件時間點，默認凌晨 4點  deleteWhen=04  #文件保留時間，默認 48 小時  fileReservedTime=120  #commitLog每個文件的大小默認1G  mapedFileSizeCommitLog=1073741824  #ConsumeQueue每個文件默認存30W條，根據業務情況調整  mapedFileSizeConsumeQueue=300000  #檢測物理文件磁碟空間  diskMaxUsedSpaceRatio=88  #存儲路徑  storePathRootDir=/opt/install/rocketmq/store  #commitLog 存儲路徑  storePathCommitLog=/opt/install/rocketmq/store/commitlog  #消費隊列存儲路徑存儲路徑  storePathConsumeQueue=/opt/install/rocketmq/store/consumequeue  #消息索引存儲路徑  storePathIndex=/opt/install/rocketmq/store/index  #checkpoint 文件存儲路徑  storeCheckpoint=/opt/install/rocketmq/store/checkpoint  #abort 文件存儲路徑  abortFile=/opt/install/rocketmq/store/abort  #限制的消息大小  maxMessageSize=65536  #刷盤方式  flushDiskType=ASYNC_FLUSH
# 啟動broker node02  nohup sh /opt/install/rocketmq/bin/mqbroker -c /opt/install/rocketmq/conf/2m-noslave/node02.properties >/dev/null 2>&1 &  # 啟動broker node03  nohup sh /opt/install/rocketmq/bin/mqbroker -c /opt/install/rocketmq/conf/2m-noslave/node03.properties >/dev/null 2>&1 &
# rocketmq 控制台安裝  # 註：需要安裝git、maven  git clone https://github.com/apache/rocketmq-externals.git  cd rocketmq-externals-maste/rocketmq-console/src/main/resources/  vim application.properties  # 設置後台訪問埠  server.port=8080  # nameserver配置  rocketmq.config.namesrvAddr=10.255.20.147:9876  cd  rocketmq-externalsrocketmq-console  # maven打包  mvn clean package -Dmaven.test.skip=true  前台啟動：  java -jar rocketmq-console-ng-1.0.1.jar  後台啟動：  nohup java -jar rocketmq-console-ng-1.0.0.jar --server.port=8080 --rocketmq.config.namesrvAddr=10.255.20.147:9876 >/dev/null 2>&1 &

02

—

RockMQ架構

RocketMQ是一個隊列模型的消息中間件，具有高性能、高可靠、高實時、分散式特點。

由四個組成部分：

NameServer：提供輕量級的服務發現和路由。每個 NameServer 記錄完整的路由資訊，提供相應的讀寫服務，支援快速的存儲擴展；

  • Broker 管理：NameServer 接受 Broker 群集的註冊，並提供心跳機制以檢查 Broker 是否還存活
  • 路由管理：每個 NameServer 保存 Broker 群集的完整路由資訊以及 客戶端（producer和consumer）查詢的 Queue 資訊

Broker：提供輕量級的 TOPIC 和 QUEUE 機制來照顧消息存儲。它們支援「推和拉」模型，包含容錯機制（2個或3個副本），並提供強大的峰值填充功能和按其原始時間順序累積數千億條消息的能力。此外，代理提供故障恢復，豐富的指標統計資訊和警報機制；

  • 負責消息的存儲和傳遞，消息查詢，HA保證
    • 遠程處理模組：處理客戶端對broker的請求；
    • 客戶端管理：管理客戶端（producer和consumer），維護消費者的 topic 訂閱；
    • 存儲服務：在物理磁碟上存儲和查詢消息；
    • 高可用服務：在Master和Slave之間建立消息同步機制；
    • 索引服務：通過指定key為消息建立索引，提供快速查詢消息；

每個 Broker 與 Name Server 集群中的所有節點建立長連接，定時註冊 Topic 資訊到所有Name Server；

Broker 分為 Master 與 Slave；

Master 與 Slave 的對應關係通過指定相同的BrokerName，不同的BrokerId 來定義，BrokerId 為0 表示Master，非0 表示Slave；

Producer：支援分散式部署。分散式生產者通過多種負載均衡模式將消息發送到Broker集群。發送過程支援快速失敗（如何避免消息丟失）並且延遲低。

  • Producer 與 Name Server 集群中的其中一個節點（隨機選擇）建立長連接，拿Topic資訊；
  • Producer 向提供Topic 服務的Master 建立長連接，且定時向Master 發送心跳；

Consumer：消費者也支援Push和Pull模型中的分散式部署。它還支援集群使用和消息廣播。它提供了實時消息訂閱機制，可以滿足大多數消費者的需求。

  • Consumer 與Name Server 集群中的其中一個節點（隨機選擇）建立長連接，拿Topic資訊；
  • Consumer 向提供Topic服務的Master、Slave建立長連接，且定時向Master、Slave發送心跳；
  • Consumer既可以從Master訂閱消息，也可以從Slave訂閱消息，訂閱規則由Broker配置決定；

Topic是生產者在發送消息和消費者在拉取消息的類別。Topic與生產者和消費者之間的關係非常鬆散。具體來說，一個Topic可能有0個，一個或多個生產者向它發送消息；相反，一個生產者可以發送不同類型Topic的消息。類似的，消費者組可以訂閱一個或多個主題，只要該組的實例保持其訂閱一致即可。

Tag，換句話的意思就是子主題，為用戶提供了額外的靈活性。有了Tag，來自同一業務模組的具有不同目的的消息可以具有相同的主題和不同的標記。標籤有助於保持程式碼的清晰和連貫，同時標籤也方便RocketMQ提供的查詢功能。

GroupName，代表具有相同角色的生產者組合或消費者組合，稱為生產者組或消費者組。

  • 作用是在集群HA的情況下，一個生產者down之後，本地事務回滾後，可以繼續聯繫該組下的另外一個生產者實例，不至於導致業務走不下去。在消費者組中，可以實現消息消費的負載均衡和消息容錯目標。
  • 另外，有了GroupName，在集群下，動態擴展容量很方便。只需要在新加的機器中，配置相同的GroupName。啟動後，就立即能加入到所在的群組中，參與消息生產或消費。

Queue，是數據分片的產物，首先，一個Topic分布在一個Broker上的子集定義為一個Topic分片，將每個分片在劃分成多個Queue，每個Topic分片等分的Queue的數量可以不同，由用戶在創建Topic時指定。

RocketMQ的負載均衡策略規定：Consumer數量應該小於等於Queue數量，如果Consumer超過Queue數量，那麼多餘的Consumer 將不能消費消息。

在一個Consumer Group內，Queue和Consumer之間的對應關係是一對多的關係：一個Queue最多只能分配給一個Consumer，一個Cosumer可以分配得到多個Queue。這樣的分配規則，每個Queue只有一個消費者，可以避免消費過程中的多執行緒處理和資源鎖定，有效提高各Consumer消費的並行度和處理效率。

Queue是Topic在一個Broker上的分片等分為指定份數後的其中一份，是負載均衡過程中資源分配的基本單元。

03

—

關鍵特性

消息持久化

  • 所有數據單獨存儲到一個CommitLog，完全順序寫，隨機讀。
  • 對最終用戶展現的隊列實際只存儲消息在Commit Log 的位置資訊，並且串列方式刷盤。

刷盤策略

RocketMQ的所有消息都是持久化的，先寫入系統PAGECACHE，然後刷盤，可以保證記憶體與磁碟都有一份數據，訪問時，直接從記憶體讀取。

順序消息查詢

可以根據key查詢和根據msgId查詢 順序消息

MQ提供一種嚴格按照順序進行發布和消費的消息類型。順序消息由兩個部分組成：順序發布和順序消費。

順序消息包含兩種類型：

分區順序：一個Partition內所有的消息按照先進先出的順序進行發布和消費；

全局順序：一個Topic內所有的消息按照先進先出的順序進行發布和消費；

以上是RocketMQ的順序消息的實現，將不同訂單路由到不同的分區中，而消費端一個分區只有一個執行緒進行消費，從而保證消費的順序。不過，也存在一些缺陷，如下：

1、發送順序消息無法利用集群 FailOver 特性；

2、消費順序消息的並行度依賴於隊列數量；

3、隊列熱點問題，個別隊列由於哈希不均導致消息過多，消費速度跟不上，產生消息堆積問題；

4、遇到消息失敗的消息，無法跳過，當前隊列消費暫停；

發送消息負載均衡

訂閱消息負載均衡

並行消費：指定滑動窗口，可以有多個執行緒並行消費； 消息重試：默認16次，進入死信隊列； HA，Master和Slave之間複製，支援同步、非同步方式；

  1. http://rocketmq.apache.org/
  2. RocketMQ-原理簡介
  3. https://www.jianshu.com/p/e30abf2753cf
  4. https://blog.csdn.net/quhongwei_zhanqiu/article/details/39153195
  5. https://www.cnblogs.com/hzmark/p/orderly_message.html

Tags:

VirMach 便宜 VPS