a视频在线观看,在线一级片,涩涩视频网

概述&選型
單機安裝配置
雙機主從高可用搭建
- 啟動多個NameServer 和 Broker
- 重要參數(shù)說明
可視化管理平臺
SpringBoot整合RocketMQ
- 引入組件rocketmq-spring-boot-starter 依賴
- 修改application.yml，添加RocketMQ相關(guān)配置
- 編寫消息生產(chǎn)者 MessageProduce
- 編寫消息消費者 MessageConsumer
- 編寫單元測試發(fā)送消息
- 測試

概述&選型

消息隊列作為高并發(fā)系統(tǒng)的核心組件之一，能夠幫助業(yè)務(wù)系統(tǒng)解構(gòu)提升開發(fā)效率和系統(tǒng)穩(wěn)定性。主要用于三種典型場景：應(yīng)用解耦、流量消峰、消息分發(fā)。

目前主流的MQ主要是Rocketmq、kafka、Rabbitmq，Rocketmq相比于Rabbitmq、kafka具有主要優(yōu)勢特性有：

支持事務(wù)型消息（消息發(fā)送和DB操作保持兩方的最終一致性，rabbitmq和kafka不支持）
支持結(jié)合rocketmq的多個系統(tǒng)之間數(shù)據(jù)最終一致性（多方事務(wù)，二方事務(wù)是前提）
支持18個級別的延遲消息（rabbitmq和kafka不支持）
支持指定次數(shù)和時間間隔的失敗消息重發(fā)（kafka不支持，rabbitmq需要手動確認）
支持consumer端tag過濾，減少不必要的網(wǎng)絡(luò)傳輸（rabbitmq和kafka不支持）
支持重復(fù)消費（rabbitmq不支持，kafka支持）

本文主要介紹RocketMQ的單機安裝、雙機主從高可用安裝配置、運維管理平臺搭建、與SpringBoot整合幾個知識點，具備相關(guān)知識技能的同學(xué)請直接拉到最后點個 “在看” 即可。

文章開始之前需要先準(zhǔn)備好JDK1.8或以上的服務(wù)器環(huán)境以及從rocketmq官網(wǎng)下載好二進制安裝包，下載地址http://rocketmq.apache.org/dowloading/releases/

單機安裝配置

工欲善其事必先利其器，要想深入了解RocketMQ得先把環(huán)境安裝好，咱們先開始單機版RocketMQ的安裝！

解壓安裝
unzip rocketmq-all-4.7.0-bin-release.zip

啟動 Name Server
> nohup sh bin/mqnamesrv &

查看 Name Server啟動日志
> tail -f ~/logs/rocketmqlogs/namesrv.log

微服務(wù)架構(gòu)設(shè)計RocketMQ基礎(chǔ)及環(huán)境整合

啟動 Broker Server
> nohup sh bin/mqbroker -n localhost:9876 &

查看 Broker Server 啟動日志
> tail -f ~/logs/rocketmqlogs/broker.log

微服務(wù)架構(gòu)設(shè)計RocketMQ基礎(chǔ)及環(huán)境整合

單機情況下安裝使用RocketMQ很簡單，只需要分別啟動NameServer和Broker Server即可！

關(guān)閉RockerMQ需要使用下面的命令：

# 先關(guān)閉Broker Server> sh bin/mqshutdown broker
# 再關(guān)閉NameServer> sh bin/mqshutdown namesrv

雙機主從高可用搭建

為了消除單機故障，增加可靠性或增大吞吐量，可以在多臺服務(wù)器上部署多個NameServer和Broker，并為每個Broker部署一個或多個Slave。本節(jié)將說明使用兩臺機器，搭建雙主、雙從、無單點故障的高可用RocketMQ集群。假設(shè)現(xiàn)在有兩臺服務(wù)器，IP地址分別為：192.168.100.43和192.168.100.44，部署架構(gòu)如下：

微服務(wù)架構(gòu)設(shè)計RocketMQ基礎(chǔ)及環(huán)境整合

啟動多個NameServer 和 Broker

首先需要在兩臺服務(wù)器上分別啟動NameServer（nohup sh bin/mqnamesrv &），這樣我們就得到了一個無單點的NameServer服務(wù)，服務(wù)地址為192.168.100.43:9876和192.168.100.44:9876。

然后在兩臺服務(wù)器中RocketMQ的conf目錄分別建立兩個文件 broker-master.properties，broker-slave.properties，下面是不同服務(wù)器的配置說明：

192.168.100.43 機器上的broker-master.properties文件：

namesrvAddr = 192.168.100.43:9876;192.168.100.44:9876
brokerClusterName = DefaultCluster
brokerName = broker-a
brokerId = 0
deleteWhen = 04
fileReservedTime = 48
brokerRole = SYNC_MASTER
flushDiskType = ASYNC_FLUSH
listenPort = 10911
storePathRootDir = /app/rocketmq/store-a

192.168.100.43 機器上的broker-slave.properties文件：

namesrvAddr = 192.168.100.43:9876;192.168.100.44:9876
brokerClusterName = DefaultCluster
brokerName = broker-b
brokerId = 1
deleteWhen = 04
fileReservedTime = 48
brokerRole = SLAVE
flushDiskType = ASYNC_FLUSH
listenPort = 11011
storePathRootDir = /app/rocketmq/store-b

192.168.100.44 機器上的broker-master.properties文件：

namesrvAddr = 192.168.100.43:9876;192.168.100.44:9876
brokerClusterName = DefaultCluster
brokerName = broker-b
brokerId = 0
deleteWhen = 04
fileReservedTime = 48
brokerRole = SYNC_MASTER
flushDiskType = ASYNC_FLUSH
listenPort = 10911
storePathRootDir = /app/rocketmq/store-b

192.168.100.44 機器上的broker-slave.properties文件：

namesrvAddr = 192.168.100.43:9876;192.168.100.44:9876
brokerClusterName = DefaultCluster
brokerName = broker-a
brokerId = 1
deleteWhen = 04
fileReservedTime = 48
brokerRole = SLAVE
flushDiskType = ASYNC_FLUSH
listenPort = 11011
storePathRootDir = /app/rocketmq/store-a

然后分別使用如下命令啟動兩臺服務(wù)器的主節(jié)點和從節(jié)點

nohup sh bin/mqbroker -c conf/broker-master.properties &
nohup sh bin/mqbroker -c conf/broker-slave.properties &

這樣一個高可用的RockerMQ集群就搭建好了，我們登陸可視化運維管理界面查看集群狀態(tài)，集群正常啟動。

微服務(wù)架構(gòu)設(shè)計RocketMQ基礎(chǔ)及環(huán)境整合

重要參數(shù)說明

本節(jié)主要是對Broker的配置文件中用到的參數(shù)進行說明

namesrvAddr = 192.168.100.43:9876;192.168.100.44:9876
指定NameServer的地址，可以是多個。

brokerClusterName = DefaultCluster
Cluster地址，如果集群數(shù)量比較多，可以分成多個Cluster，每個Cluster供一個業(yè)務(wù)群使用。

brokerName = broker-a
Broker的名稱，Master 和Slave 通過使用相同的 Broker 名稱來表明相互關(guān)系，以說明某個Slave 是哪個Master 的 Slave。

brokerId = 1
一個Master可以有多個Slave，0表示Master，大于0的表示不同Slave的ID。

fileReservedTime = 48
在磁盤上保存消息的時長，單位是小時，自動刪除超時的消息。

deleteWhen = 04
與 fileReservedTime 參數(shù)對應(yīng)，表明在幾點做消息刪除動作，默認是凌晨4點。

brokerRole = SYNC_MASTER
brokerRole的可選參數(shù)有SYNC_MASTER，ASYNC_MASTER，SLAVE三種。SYNC 和ASYNC 表示MASTER 和SLAVE 之間同步消息的機制，SYNC的意思是當(dāng)Slave 和 Master 的消息同步完成后再返回發(fā)送成功的狀態(tài)。

flushDiskType = ASYNC_FLUSH
flushDiskType 表示刷盤策略，可選值有ASYNC_FLUSH 和 SYNC_FLUSH兩種，分別代表同步刷盤和異步刷盤。同步情況下，消息只有真正寫入磁盤才返回成功狀態(tài)；異步情況下，消息寫入page_cache后就返回成功狀態(tài)。

listenPort = 11011
Broker監(jiān)聽的端口，一臺服務(wù)器啟動多個Broker，需要設(shè)置不同的監(jiān)聽端口避免端口沖突。

storePathRootDir = /app/rocketmq/store-a
存儲消息以及配置信息的根目錄。

可視化管理平臺

RocketMQ可以使用rocketmq-externals作為運維管理平臺，Github地址https://github.com/apache/rocketmq-externals，我們需要將源碼下載下來后再進行手動編譯，過程如下：

下載
從github（https://github.com/apache/rocketmq-externals）下載RocketMQ可視化管理工具 rocketmq-externals 的源碼；

打包
下載完成后切換進rocketmq-console目錄，使用maven命令對其打包 mvn clean package -Dmaven.test.skip=true
打包完成后生成可執(zhí)行文件rocketmq-console-ng-1.0.1.jar

運行
使用 java -jar rocketmq-console-ng-1.0.1.jar --server.port=8080 --rocketmq.config.namesrvAddr=xxxx.xxx.xxx.xxx:9876 命令啟動

這里注意需要設(shè)置兩個參數(shù)：
--server.port 為運行的這個web應(yīng)用的端口，如果不設(shè)置的話默認為8080；
--rocketmq.config.namesrvAddr 為RocketMQ命名服務(wù)地址，若NameServer為集群則使用英文 ; 分割

訪問
瀏覽器訪問 xxx.xxx.xxx.xxx:8080 進入控制臺界面，效果如下

微服務(wù)架構(gòu)設(shè)計RocketMQ基礎(chǔ)及環(huán)境整合

SpringBoot整合RocketMQ

在SpringBoot中整合RocketMQ主要用到 rocketmq-spring-boot-starter 組件，下面是詳細整合過程。

引入組件rocketmq-spring-boot-starter 依賴

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

修改application.yml，添加RocketMQ相關(guān)配置

rocketmq:
  name-server: 192.168.100.43:9876;192.168.100.44:9876
  producer:
    group: test-group
    send-message-timeout: 3000

如果是集群，多個name-server使用英文 ; 分割。

編寫消息生產(chǎn)者 MessageProduce

/**
 * Description:
 * rocketMQ消息發(fā)送方法
 * @author javadaily
 */
@Component
public class MessageProduce {

    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    /**
     * 發(fā)送消息
     * @param topic 主題
     * @param message 消息體
     */
    public void sendMessage(String topic,String message){
        this.rocketMQTemplate.convertAndSend(topic,message);
    }
}

使用RocketMQTemplate發(fā)送消息

編寫消息消費者 MessageConsumer

@Slf4j
@Component
@RocketMQMessageListener(
        topic = "test-topic",
        consumerGroup = "test-group",
        selectorExpression = "*"
)
public class MessageConsumer implements RocketMQListener<String> {
    @Override
    public void onMessage(String message) {
        log.info("received message is {}", message);
    }
}

消費者只需要繼承RocketMQListener類即可，主要關(guān)注實現(xiàn)類上的 @RocketMQMessageListener 注解，配置的 topic 和 consumerGroup 需要跟消息生產(chǎn)者的配置保持一致。

編寫單元測試發(fā)送消息

@RunWith(SpringRunner.class)
@SpringBootTest
public class MessageProduceTest {
    @Autowired
    private MessageProduce messageProduce;

    @Test
    public void testSendMessage() {
        messageProduce.sendMessage("test-topic","Hello,JAVA日知錄");
    }
}