消息隊列
基本概述
MQ,Message Queue,基于TCP協(xié)議構(gòu)建的簡單協(xié)議，區(qū)別于具體的通信協(xié)議。
基于通信協(xié)議定義和抽象的更高層次的通信模型,一般都是生產(chǎn)者和消費者模型，又或者說服務端和客戶端模型。
生產(chǎn)者/消費者模型：一般通過定義生產(chǎn)者和消費者實現(xiàn)消息通信從而屏蔽復雜的底層通信協(xié)議。應用于分布式應用系統(tǒng)，而且為之提供異步解耦和削峰填谷的能力，同時也具備互聯(lián)網(wǎng)應用所需的海量堆積，高吞吐和可靠性重試機制的特性。
核心概念
- 消息主題：Message Topic,一級消息類型，生產(chǎn)者向其發(fā)送消息
- 消息生產(chǎn)者：Message Producer，消息發(fā)布者，或者消息服務終端，負責生產(chǎn)消息和發(fā)送消息到消息主題
- 消息消費者：Message Consumer，消息訂閱者，或者消息客戶終端，負責從消息主題接收并處理消費消息
- 消息實體：Message Object,消息對象，生產(chǎn)者向消息主題發(fā)送并最終傳送給消息者的數(shù)據(jù)和屬性的符號以及組合
- 消息屬性：Message Attributes,消息內(nèi)容，生產(chǎn)者對消息進行抽象和定義的相關(guān)屬性，包含Message Key 和Message Target
- 消息組：Message Group，消息分類組別，通稱一類生產(chǎn)者和消費者，通常生產(chǎn)和消費同一類消息，且消息發(fā)布和訂閱的邏輯基本一致
編程思想
數(shù)據(jù)結(jié)構(gòu)：消息隊列的數(shù)據(jù)結(jié)構(gòu)采用FIFO方式來定義與實現(xiàn)
設(shè)計模式：采用觀察者模式
觀察者模式：定義對象間一對多的依賴關(guān)系，當一個對象的狀態(tài)發(fā)生改變時，所有依賴于它的對象都得到通知自動更新
消息流派：
- 有Broker：基于消息中間件，所有消息通過中間件中轉(zhuǎn)，基于Broker把消息推送給消費者[消費者主動輪詢]
1.重Topic：Kafka和RocketMq,Activemq，生產(chǎn)者把消息發(fā)送Key和數(shù)據(jù)到Broker，然后由Broker比較Key之后決定推送給具體的消費者[推送消息給消費者]
2.輕Topic：Rabbitmq[AMQP],生產(chǎn)者把生產(chǎn)消息且發(fā)送Key和數(shù)據(jù)，消費者定義訂閱的隊列，Broker收到請求之后，按照一定的邏輯和規(guī)則計算出相應的Key對應的隊列，然后把消息推送給具體的消費者[消費者向生產(chǎn)者訂閱消息，消費者主動拉取消息] - 無Broker：基于Socket網(wǎng)絡(luò)[傳輸層]直連通信，代表作ZeroMQ
基本概述
NameServer：名稱服務器[MQ命名空間服務器],大致相當于 jndi技術(shù)，更新和發(fā)現(xiàn) broker服務。用于保存Broker相關(guān)元信息，并給生產(chǎn)者和消費者查找Broker消息。每個Broker在啟動都會在名稱服務器[NameServer]注冊,生產(chǎn)者在發(fā)送消息前會根據(jù)消息主題到名稱服務器查詢獲取Broker路由消息，消費者也會定時獲取主題的路由消息。
ps：屬于無狀態(tài)服務設(shè)計，可橫向擴展，節(jié)點之間無通信，可以部署多臺機器來標識偽集群。
Broker:消息存儲中心[消息中轉(zhuǎn)角色]，負責存儲和轉(zhuǎn)發(fā)消息。接收來自生產(chǎn)者的消息并進行存儲，消費者從這拉取消息。存儲與消息相關(guān)的元數(shù)據(jù)，主要包括用戶組,消息進度偏移量，隊列消息等。其中Broker分為Master和Slave節(jié)點：
- Master節(jié)點：可讀可寫
- Slave節(jié)點：只可讀不可寫
其部署方式：
- 單機Master:Broker重啟和宕機之后服務不可用,不建議生產(chǎn)使用
- 多機Master:所有消息服務器均是Master,沒有Slave。
優(yōu)點：配置簡單，單個Master重啟和宕機維護對應用無影響
缺點：單機重啟和宕機期間，機器上未被消費的消息在機器恢復期間不可訂閱，消息實時性會受影響 - 多機Master多機Slave[同步雙寫]:
優(yōu)點：數(shù)據(jù)和服務無單點故障，Master重啟和宕機消息無延遲,數(shù)據(jù)和服務可用高
缺點：相對異步復制性能較低，發(fā)送消息的延時較高 - 多機Master多機Slave[異步復制]:
優(yōu)點：消息丟失較少，實時性高，Master重啟和宕機可繼續(xù)從Slave消費
缺點：Master重啟和宕機后在磁盤損壞會導致消息丟失，但是情況比較少
ps：一個Master可以對應多個Slave，但是一個Slave只能對應一個Master。Master和Slave的對應關(guān)系通過指定相同的BrokerName，不同的BrokerId來定義。BrokerId為0表示Master，BrokerId非0表示Slave。然后所有的Broker和Name Server上的節(jié)點建立長連接,定時注冊Topic信息到所有Name Server。
Producer：消息生產(chǎn)者->負責生產(chǎn)消息,生產(chǎn)者向消息服務器發(fā)送業(yè)務應用程序生成的消息。主要有同步發(fā)送和異步發(fā)送方式兩種，其中：
- 同步發(fā)送：消息發(fā)送方發(fā)出數(shù)據(jù)后，在消息接收方發(fā)送響應之后再發(fā)送下一個數(shù)據(jù)。一般適用于重要消息通知場景[重要通知郵件，營銷短信推送]
- 異步發(fā)送：消息發(fā)送方發(fā)出數(shù)據(jù)后，不用等待接收方發(fā)回響應，接著發(fā)送下一個數(shù)據(jù)。一般適用于鏈路耗時較長，而對響應時間敏感的業(yè)務場景[視頻上傳通知 啟動自動轉(zhuǎn)碼服務處理通知]
- 單向發(fā)送：負責發(fā)送消息而不等待服務器回應且沒有回調(diào)函數(shù)觸發(fā)。一般適用于對可靠性要求不高的業(yè)務場景[日志收集]
ps：Producer與Name Server其中一個節(jié)點建立連接。定期從Name Server取Topic信息。并與提供該Topic信息的Master建立長連接。Producer也可以集群部署。
Consumer：消息消費者 負責消費消息，從消息服務器拉取消息并將其輸入用戶應用程序中。主要分為拉取型消費者和推送型消費者：
拉取型消費者：Pull Consumer->主動從消息服務器拉取消息，只要批量拉取消息,用戶就會啟動消費過程
推送型消費者：Push Consumer->封裝消息的拉取,消費進度和其它內(nèi)部維護工作，消息到達之后便執(zhí)行回調(diào)接口留給用戶應用程序來實現(xiàn)。屬于被動消費類型，Push拉取時需要注冊消息費者監(jiān)聽器，當監(jiān)聽器被觸發(fā)之后開始消費消息。
ps：Consumer 與Name Server 集群中的其中一個節(jié)點（隨機選擇）建立長連接，定期從Name Server 取Topic 路由信息，并向提供Topic服務的Master、Slave建立長連接，且定時向Master、Slave發(fā)送心跳。Consumer既可以從Master訂閱消息，也可以從Slave訂閱消息，訂閱規(guī)則由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.執(zhí)行：docker pull foxiswho/rocketmq
3.創(chuàng)建docker存儲根目錄并且授權(quán):
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：創(chuàng)建/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.創(chuàng)建/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
# 如果是本地程序調(diào)用云主機 mq，這個需要設(shè)置成 云主機 IP
brokerIP1=Server-IP
#限制的消息大小
maxMessageSize=65536
# 檢測物理文件磁盤空間
diskMaxUsedSpaceRatio=88
mapedFileSizeCommitLog=1073741824
mapedFileSizeConsumeQueue=50000000
#并發(fā)send線程數(shù)，多線程來發(fā)送消息可能會出現(xiàn)broker busy
sendMessageThreadPoolNums=128
useReentrantLockWhenPutMessage=true
#在發(fā)送消息時，自動創(chuàng)建服務器不存在的topic，默認創(chuàng)建的隊列數(shù)
defaultTopicQueueNums=8
highSpeedMode=false
commercialBaseCount=1
maxErrorRateOfBloomFilter=20
accessMessageInMemoryMaxRatio=40
#無讀寫客戶端存活時間
clientChannelMaxIdleTimeSeconds=120
flushDelayOffsetInterval=10000
serverSocketRcvBufSize=131072
#單次 Pull 消息(內(nèi)存)傳輸?shù)?最大字節(jié)數(shù)
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.創(chuàng)建/docker/rocketmq/console/data目錄
2.Server-IP為broker.conf配置brokerIP1值
7.最終部署結(jié)果：
整合Rocketmq開發(fā)實戰(zhàn)
1.配置Rocketmq的Maven依賴：
<!-- https://mvnrepository.com/artifact/org.apache.rocketmq/rocketmq-client -->
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>4.4.0</version>
</dependency>
ps：
1.目前rocketmq最新版本是4.7.1：
<!-- https://mvnrepository.com/artifact/org.apache.rocketmq/rocketmq-client -->
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>4.7.1</version>
</dependency>
2.這里選擇的是rocketmq版本[4.4.0],因為部署是4.4.0版本，推薦部署版本相同的rocketmq
2.創(chuàng)建pivotal-cloud-queue工程:
ps：創(chuàng)建消息隊列統(tǒng)一模塊工程，在業(yè)務應用程序服務模塊依賴該工程，并封裝消息生產(chǎn)者[Producer]和消息消費者[Consumer]處理類。
3.封裝rocketmq屬性配置類:
添加spring-boot-configuration-processor依賴:
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>1.18.12</version>
<scope>provided</scope>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-configuration-processor</artifactId>
<optional>true</optional>
</dependency>
編寫RocketmqProperties屬性配置類:
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項目系統(tǒng)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消息隊列生產(chǎn)者-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 {
}
}
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生產(chǎn)者模板類:
package com.pivotal.cloud.queue.template;
/**
* @className: com.pivotal.cloud.queue.template.RocketmqTemplate
* @title: RocketmqTemplate
* @description: 封裝Pivotal項目RocketmqTemplate類
* @content: PivotalCloud項目系統(tǒng)RocketmqTemplate生產(chǎn)者模板類
* @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消費者監(jiān)聽器類:
package com.pivotal.cloud.queue.listener;
/**
* @className: com.pivotal.cloud.queue.listener.RocketmqListener
* @title: RocketmqListener
* @description: 封裝Pivotal項目RocketmqListener類
* @content: PivotalCloud項目系統(tǒng)RocketmqListener消費者監(jiān)聽器類
* @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工廠類:
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,
版權(quán)聲明：本文為博主原創(chuàng)文章，遵循相關(guān)版權(quán)協(xié)議，如若轉(zhuǎn)載或者分享請附上原文出處鏈接和鏈接來源。
