1. 中間件下載

官網(wǎng)上直接下載毅臊，比較簡單就不贅述！

2. 簡介

2.1. RocketMQ是什么

RocketMQ

RocketMQ 是一個隊列模型的消息中間件黑界，具有高性能管嬉、高可靠、高實時朗鸠、分布式特點蚯撩，具體特性如下：

Producer、Consumer烛占、隊列都可以分布式胎挎。
Producer 向隊列輪流發(fā)送消息，隊列集合稱為Topic忆家，Consumer 如果做廣播消費犹菇，則一個consumer實例消費這個Topic對應的所有隊列，如果做集群消費芽卿，則多個Consumer 實例平均消費這個topic 對應的隊列集合揭芍。
能夠保證嚴格的消息順序
提供豐富的消息拉取模式
高效的訂閱者水平擴展能力
實時的消息訂閱機制
億級消息堆積能力
較少的依賴

2.2. 概念

2.2.1. 消息模型（Message Model）

RocketMQ主要由 Producer、Broker卸例、Consumer 三部分組成称杨，其中Producer 負責生產(chǎn)消息，Consumer 負責消費消息币厕，Broker 負責存儲消息列另。Broker 在實際部署過程中對應一臺服務器，每個 Broker 可以存儲多個Topic的消息旦装，每個Topic的消息也可以分片存儲于不同的 Broker页衙。Message Queue 用于存儲消息的物理地址，每個Topic中的消息地址存儲于多個 Message Queue 中阴绢。ConsumerGroup 由多個Consumer 實例構(gòu)成店乐。

2.2.2. 消息生產(chǎn)者（Producer）

負責生產(chǎn)消息，一般由業(yè)務系統(tǒng)負責生產(chǎn)消息呻袭。一個消息生產(chǎn)者會把業(yè)務應用系統(tǒng)里產(chǎn)生的消息發(fā)送到broker服務器眨八。RocketMQ提供多種發(fā)送方式，同步發(fā)送左电、異步發(fā)送廉侧、順序發(fā)送页响、單向發(fā)送。同步和異步方式均需要Broker返回確認信息段誊，單向發(fā)送不需要闰蚕。

2.2.3. 消息消費者（Consumer）

負責消費消息，一般是后臺系統(tǒng)負責異步消費连舍。一個消息消費者會從Broker服務器拉取消息没陡、并將其提供給應用程序。從用戶應用的角度而言提供了兩種消費形式：拉取式消費索赏、推動式消費盼玄。

2.2.4. 主題（Topic）

表示一類消息的集合，每個主題包含若干條消息潜腻，每條消息只能屬于一個主題埃儿，是RocketMQ進行消息訂閱的基本單位。

2.2.5. 名字服務（Name Server）

名稱服務充當路由消息的提供者融涣。生產(chǎn)者或消費者能夠通過名字服務查找各主題相應的Broker IP列表蝌箍。多個Namesrv實例組成集群，但相互獨立暴心，沒有信息交換。

2.2.6. 拉取式消費（Pull Consumer）

Consumer消費的一種類型杂拨，應用通常主動調(diào)用Consumer的拉消息方法從Broker服務器拉消息专普、主動權(quán)由應用控制。一旦獲取了批量消息弹沽，應用就會啟動消費過程檀夹。

2.2.7. 推動式消費（Push Consumer）

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

2.2.8. 生產(chǎn)者組（Producer Group）

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

2.2.9. 消費者組（Consumer Group）

同一類Consumer的集合沛婴，這類Consumer通常消費同一類消息且消費邏輯一致吼畏。消費者組使得在消息消費方面，實現(xiàn)負載均衡和容錯的目標變得非常容易嘁灯。要注意的是泻蚊，消費者組的消費者實例必須訂閱完全相同的Topic。RocketMQ 支持兩種消息模式：集群消費（Clustering）和廣播消費（Broadcasting）丑婿。

2.2.10. 普通順序消息（Normal Ordered Message）

普通順序消費模式下性雄，消費者通過同一個消費隊列收到的消息是有順序的没卸，不同消息隊列收到的消息則可能是無順序的。

2.2.11. 嚴格順序消息（Strictly Ordered Message）

嚴格順序消息模式下秒旋，消費者收到的所有消息均是有順序的约计。

2.3. 架構(gòu)設(shè)計

2.3.1. 技術(shù)架構(gòu)

技術(shù)架構(gòu)圖

2.3.2. 部署架構(gòu)

部署架構(gòu)圖

3. 安裝

由于RocketMQ安裝比較簡單，在單機這塊就省略滩褥。

3.1. 單機

略２◎取！９寮濉铺然！

3.2. 集群

3.2.1. 集群的方式

單 Master：這種方式風險較大，一旦Broker重啟或者宕機時酒甸，會導致整個服務不可用魄健，不建議線上環(huán)境使用。
多 Master 模式：一個集群無 Slave插勤，全是 Master沽瘦，例如 2 個 Master 或者 3 個 Master
- 優(yōu)點：配置簡單，單個Master 宕機或重啟維護對應用無影響农尖，在磁盤配置為 RAID10 時析恋，即使機器宕機不可恢復情況下，由與 RAID10磁盤非呈⒖ǎ可靠助隧，消息也不會丟（異步刷盤丟失少量消息，同步刷盤一條不丟）滑沧。性能最高并村。
- 缺點：單臺機器宕機期間，這臺機器上未被消費的消息在機器恢復之前不可訂閱滓技，消息實時性會受到受到影響哩牍。
多 Master 多 Slave 模式，異步復制：每個 Master 配置一個 Slave令漂，有多對Master-Slave膝昆，HA
采用異步復制方式，主備有短暫消息延遲叠必，毫秒級外潜。
- 優(yōu)點：即使磁盤損壞，消息丟失的非常少挠唆，且消息實時性不會受影響处窥，因為Master 宕機后，消費者仍然可以從 Slave玄组。
- 缺點：Master 宕機滔驾，磁盤損壞情況谒麦，會丟失少量消息。
多 Master 多 Slave 模式哆致，同步雙寫：每個 Master 配置一個 Slave绕德，有多對Master-Slave，HA
采用同步雙寫方式摊阀，主備都寫成功耻蛇，向應用返回成功。
- 優(yōu)點：數(shù)據(jù)與服務都無單點胞此，Master宕機情況下臣咖，消息無延遲，服務可用性與數(shù)據(jù)可用性都非常高漱牵。
- 缺點：性能比異步復制模式略低夺蛇，大約低 10%左右，發(fā)送單個消息的 RT 會略高酣胀。目前主宕機后刁赦，備機不能自動切換為主機，后續(xù)會支持自動切換功能

后兩種的方式比較復雜闻镶，涉及主從同步的問題甚脉，非必要的場景，建議采用多Master 這種方式铆农。而我下面的例子也是以多Master來部署的宦焦。

3.2.2. 部署

服務器環(huán)境準備

序號	IP	角色	模式
A	192.168.244.128	nameServer1,brokerServer1	Master1
B	192.168.244.129	nameServer2,brokerServer2	Master2

修改HOST

在 A、B 兩服務器中將HOST文件修改顿涣，vi /etc/hosts

修改內(nèi)容如下：

192.168.244.128 rocketmq-nameserver1
192.168.244.128 rocketmq-master1

192.168.244.129 rocketmq-nameserver2
192.168.244.129 rocketmq-master2

上傳文件

在A、B兩臺機器上傳RockerMQ文件酝豪，并解壓涛碑。我這里上傳和解壓路徑在/usr/local

[root@centos8 rocketmq]$ pwd
/usr/local/rocketmq
[root@centos8 rocketmq]$ ll
total 40
drwxr-xr-x. 2 root root    83 Jun 24 02:49 benchmark
drwxr-xr-x. 3 root root  4096 Jun 24 02:02 bin
drwxr-xr-x. 6 root root   211 Jun  2 02:09 conf
drwxr-xr-x. 2 root root  4096 Jun 24 02:49 lib
-rw-r--r--. 1 root root 17336 Jun  2 02:09 LICENSE
-rw-r--r--. 1 root root  1338 Jun  2 02:09 NOTICE
-rw-r--r--. 1 root root  5069 Jun 24 02:02 README.md
[root@centos8 rocketmq]$

創(chuàng)建存儲路徑

在A、B兩臺機器執(zhí)行創(chuàng)建路徑孵淘。

[root@centos8 rocketmq]$ mkdir /usr/local/rocketmq/store
[root@centos8 rocketmq]$ mkdir /usr/local/rocketmq/store/commitlog
[root@centos8 rocketmq]$ mkdir /usr/local/rocketmq/store/consumequeue
[root@centos8 rocketmq]$ mkdir /usr/local/rocketmq/store/index

修改配置文件

這里貼一個標準配置文件蒲障，具體如下：


// 所屬集群名字
brokerClusterName=rocketmq-cluster
//  broker名字，注意此處不同的配置文件填寫的不一樣
brokerName=broker-a/broker-b ## 需要按機器A/B 區(qū)分
// 0 表示 Master瘫证，>0 表示 Slave
brokerId=0
//  nameServer地址揉阎，分號分割

namesrvAddr=rocketmq-nameserver1:9876;rocketmq-nameserver2:9876
// 在發(fā)送消息時，自動創(chuàng)建服務器不存在的topic背捌，默認創(chuàng)建的隊列數(shù)
defaultTopicQueueNums=4
// 是否允許 Broker 自動創(chuàng)建Topic毙籽，建議線下開啟，線上關(guān)閉
autoCreateTopicEnable=true
// 是否允許 Broker 自動創(chuàng)建訂閱組毡庆，建議線下開啟坑赡，線上關(guān)閉
autoCreateSubscriptionGroup=true
// Broker 對外服務的監(jiān)聽端口
listenPort=10911
// 刪除文件時間點烙如，默認凌晨 4點
deleteWhen=04
// 文件保留時間，默認 48 小時
fileReservedTime=120
// commitLog每個文件的大小默認1G
mapedFileSizeCommitLog=1073741824
//  ConsumeQueue每個文件默認存30W條毅否，根據(jù)業(yè)務情況調(diào)整
mapedFileSizeConsumeQueue=300000
// destroyMapedFileIntervalForcibly=120000
// redeleteHangedFileInterval=120000
// 檢測物理文件磁盤空間
diskMaxUsedSpaceRatio=88
// 存儲路徑
storePathRootDir=/usr/local/rocketmq/store
// commitLog 存儲路徑
storePathCommitLog=/usr/local/rocketmq/store/commitlog
// 消費隊列存儲路徑存儲路徑
storePathConsumeQueue=/usr/local/rocketmq/store/consumequeue
// 消息索引存儲路徑
storePathIndex=/usr/local/rocketmq/store/index
// checkpoint 文件存儲路徑
storeCheckpoint=/usr/local/rocketmq/store/checkpoint
// abort 文件存儲路徑
abortFile=/usr/local/rocketmq/store/abort
// 限制的消息大小
maxMessageSize=65536
// flushCommitLogLeastPages=4
// flushConsumeQueueLeastPages=2
// flushCommitLogThoroughInterval=10000
// flushConsumeQueueThoroughInterval=60000
// Broker 的角色
//  - ASYNC_MASTER 異步復制Master
//  - SYNC_MASTER 同步雙寫Master
//  - SLAVE
brokerRole=ASYNC_MASTER
//  刷盤方式
//  - ASYNC_FLUSH 異步刷盤
//  - SYNC_FLUSH 同步刷盤
flushDiskType=ASYNC_FLUSH
//  checkTransactionMessageEnable=false

//  發(fā)消息線程池數(shù)量
//  sendMessageThreadPoolNums=128
//  拉消息線程池數(shù)量
//  pullMessageThreadPoolNums=128

機器A

[root@centos8 rocketmq]$ vi/usr/local/rocketmq/conf/2m-noslave/broker-a.properties

其中brokerName=broker-a

在機器B

[root@centos8 rocketmq]$ vi/usr/local/rocketmq/conf/2m-noslave/broker-a.properties

其中brokerName=broker-b

修改日志配置文件

不用說肯定也是A亚铁、B兩臺都要改

[root@centos8 rocketmq]$ mkdir -p /usr/local/rocketmq/logs
[root@centos8 rocketmq]$ cd /usr/local/rocketmq/conf && sed -i 's#${user.home}#/usr/local/rocketmq#g' *.xml

修改啟動參數(shù)（A、B）

在/usr/local/rocketmq/bin 路徑下螟加，找到runbroker.sh 和 runserver.sh徘溢。

我們將這兩個文件的JAVA_OPT 參數(shù)修改下，不然默認情況下捆探，JVM配置是 8G然爆。如 JAVA_OPT="${JAVA_OPT} -server -Xms8g -Xmx8g -Xmn4g"。

修改后：

...
JAVA_OPT="${JAVA_OPT} -server -Xms1g -Xmx1g -Xmn512m"
....

啟動NameServer（A徐许、B）

[root@centos8 rocketmq]$ cd /usr/local/rocketmq/bin
[root@centos8 rocketmq]$ nohup sh mqnamesrv &

機器A

機器B

啟動BrokerServer（A）

[root@centos8 rocketmq]$ cd /usr/local/rocketmq/bin
[root@centos8 rocketmq]$ nohup sh mqbroker -c /usr/local/rocketmq/conf/2m-noslave/broker-a.properties &

啟動BrokerServer（B）

[root@centos8 rocketmq]$ cd /usr/local/rocketmq/bin
[root@centos8 rocketmq]$ nohup sh mqbroker -c /usr/local/rocketmq/conf/2m-noslave/broker-b.properties &

rocketmq-console安裝

rocketmq-console下載

這個包需要利用maven編譯打包施蜜。我這里打包一個，放百度云盤上雌隅，供下載翻默！

鏈接: https://pan.baidu.com/s/1hfvzJeyBG7TXnvPHtn3C5Q

提取碼: atkq

最后執(zhí)行jar文件

[root@centos8 rocketmq]$ java -jar rocketmq-console-ng-1.0.0.jar

頁面的端口是 8082，剛開始啟動有點慢恰起，稍微等會Ｐ扌怠！

rocketmq-console界面

數(shù)據(jù)清理

[root@centos8 rocketmq]$ cd /usr/local/rocketmq/bin
[root@centos8 rocketmq]$ sh mqshutdown broker
[root@centos8 rocketmq]$ sh mqshutdown namesrv

這里需要等待完全停止检盼！

[root@centos8 rocketmq]$ rm -rf /usr/local/rocketmq/store
[root@centos8 rocketmq]$ mkdir /usr/local/rocketmq/store
[root@centos8 rocketmq]$ mkdir /usr/local/rocketmq/store/commitlog
[root@centos8 rocketmq]$ mkdir /usr/local/rocketmq/store/consumequeue
[root@centos8 rocketmq]$ mkdir /usr/local/rocketmq/store/index

最終按照以上步驟重啟NameServer和BrokerServer即可肯污！

4. SpringBoot集成

POM文件添加依賴

<dependency>
    <groupId>org.apache.rocketmq</groupId>
    <artifactId>rocketmq-spring-boot-starter</artifactId>
    <version>{換成相應版本}</version>
</dependency>

application.yml

######### 4.1. RocketMQ ##########
rocketmq:
  name-server: 192.168.244.128:9876;192.168.244.129:9876;
  producer:
    group: drunkard
    send-message-timeout: 30000

生產(chǎn)者

@Slf4j
@RestController
public class RocketMqDemo {

    @Autowired
    RocketMQTemplate rocketMQTemplate;

    @GetMapping("send/{id}")
    public String send(@PathVariable("id") String id){
        UserVo userVo  = new UserVo(id,"侯征");
        log.warn(JSON.toJSONString(userVo));
        rocketMQTemplate.send("rocket-topic-01", MessageBuilder.withPayload(userVo).build());
        return "SUCESS";
    }
}

消費者

@Slf4j
@Component
@RocketMQMessageListener(topic = "rocket-topic-01", consumerGroup = "my-rocket-topic-01")
public class UserConsumer implements RocketMQListener<UserVo> {

    @Override
    public void onMessage(UserVo message) {
        log.warn("接受到消息: {}",message.toString());
    }
}

5. 案例下載

覺得對你有幫助，請Star

Gitee下載吨枉，希望多多給Star蹦渣。

Github下載，希望多多給Star貌亭。

RocketMQ集群以及SpringBoot集成