消息隊(duì)列 RocketMQ 入門指南
在當(dāng)今的分布式系統(tǒng)中，消息隊(duì)列（Message Queue）作為解耦、異步通信和流量削峰的重要組件，扮演著不可或缺的角色。而RocketMQ，作為阿里巴巴開源的一款高性能、高可靠的分布式消息中間件，憑借其強(qiáng)大的功能和穩(wěn)定的性能，成為了眾多開發(fā)者和企業(yè)的首選。
無論你是剛剛接觸消息隊(duì)列的新手，還是希望深入了解RocketMQ的開發(fā)者，這篇文章都將為你提供一個(gè)清晰的入門指南。我們將從RocketMQ的基本概念出發(fā)，逐步講解其核心功能，并通過簡單的實(shí)踐示例，幫助你快速上手。
一、詳解RocketMQ基礎(chǔ)概念
1. 為什么要用RocketMQ
相比于市場上的各種消息隊(duì)列，它有如下優(yōu)勢(shì):
- 性能好。
- 穩(wěn)定可靠。
- 中文社區(qū)活躍。
當(dāng)然缺點(diǎn)也是有那么一些些的，兼容性確實(shí)不太行。
2. RocketMQ優(yōu)缺點(diǎn)是什么
優(yōu)點(diǎn):
- 單機(jī)吞吐量為10w級(jí)。
- 可用性很高，支持分布式架構(gòu)。
- 擴(kuò)展性好。
- 支持10億級(jí)別的消息堆積，而且不會(huì)因?yàn)槎逊e導(dǎo)致性能下降。
- 源碼是用Java寫的，對(duì)于Java程序員來說非常方便改造。
- 參數(shù)優(yōu)化配置，消息基本可以做到0丟失。
- 使用于對(duì)可靠性要求高的金融行業(yè)。
缺點(diǎn):
- 目前只支持Java、C++客戶端，而且C++還不算完善。
- 沒有在MQ核心實(shí)現(xiàn)JMS相關(guān)接口，有些遷移改造就比較麻煩了。
3. 消息隊(duì)列使用場景
解耦: 例如用戶完成下單除了必要的庫存扣減和訂單狀態(tài)更新外，我們還需要處理一些積分系統(tǒng)、推送系統(tǒng)的無關(guān)緊要的業(yè)務(wù)處理，如果全部順序執(zhí)行，等待時(shí)間就會(huì)變得很漫長，所以我們需要借助MQ將邊角業(yè)務(wù)從業(yè)務(wù)模塊中解耦開來。
- 異步: 這點(diǎn)不必多說，上述的解耦方案就會(huì)使得積分系統(tǒng)、促銷系統(tǒng)、推送系統(tǒng)任務(wù)異步執(zhí)行。
- 削峰: 可以理解為一個(gè)漏斗，例如我們的某個(gè)服務(wù)只能抗住10wQPS，可是當(dāng)前請(qǐng)求卻達(dá)到20w的QPS，那么我們就可以將請(qǐng)求全部先扔到MQ中，讓服務(wù)慢慢消化處理。
二、RocketMQ基礎(chǔ)安裝與實(shí)踐
1. 安裝并啟動(dòng)RocketMQ
在編寫業(yè)務(wù)代碼之前，我們必須完成一下RocketMQ的部署，首先我們自然要下載一下RocketMQ，下載地址如下，筆者下載的是rocketmq-all-4.8.0-bin-release這個(gè)版本：https://rocketmq.apache.org/download/
完成完成后，我們將其解壓到自定義的路徑，鍵入sudo vim /etc/profile配置MQ環(huán)境變量，完成后鍵入source /etc/profile使之生效，對(duì)應(yīng)的配置內(nèi)容如下所示：
export ROCKETMQ_HOME=/home/sharkchili/rocketmq-all-4.8.0-bin-release
export PATH=$PATH:$ROCKETMQ_HOME/bin需要注意的是筆者本次采用WSL的Ubuntu子系統(tǒng)時(shí)啟動(dòng)時(shí)腳本會(huì)拋出runserver.sh: 70: [[: Exec format error錯(cuò)誤，嘗試格式化和指令配置后都沒有很好的解決，于是循著報(bào)錯(cuò)找到runserver.sh這行對(duì)應(yīng)的腳本內(nèi)容，該括弧本質(zhì)上就是基于JDK內(nèi)容配置對(duì)應(yīng)的GC算法：
以筆者為里系統(tǒng)是jdk8，所以直接去掉判斷用走JDK8的配置即可：
choose_gc_options()
{
JAVA_OPT="${JAVA_OPT} -XX:+UseConcMarkSweepGC -XX:+UseCMSCompactAtFullCollection -XX:CMSInitiatingOccupancyFractinotallow=70 -XX:+CMSParallelRemarkEnabled -XX:SoftRefLRUPolicyMSPerMB=0 -XX:+CMSClassUnloadingEnabled -XX:SurvivorRatio=8 -XX:-UseParNewGC"
JAVA_OPT="${JAVA_OPT} -verbose:gc -Xloggc:${GC_LOG_DIR}/rmq_srv_gc_%p_%t.log -XX:+PrintGCDetails"
JAVA_OPT="${JAVA_OPT} -XX:+UseGCLogFileRotation -XX:NumberOfGCLogFiles=5 -XX:GCLogFileSize=30m"
}完成后鍵入./mqnamesrv &將MQ啟動(dòng)，如果彈窗輸出下面這條結(jié)果，則說明mq的NameServer啟動(dòng)成功。
Java HotSpot(TM) 64-Bit Server VM warning: Using the DefNew young collector with the CMS collector is deprecated and will likely be removed in a future release
Java HotSpot(TM) 64-Bit Server VM warning: UseCMSCompactAtFullCollection is deprecated and will likely be removed in a future release.
The Name Server boot success. serializeType=JSON然后我們?cè)冁I入./mqbroker -n 127.0.0.1:9876啟動(dòng)broker，需要注意的是默認(rèn)情況下broker占用堆內(nèi)存差不多是4g，所以讀者本地部署時(shí)建議修改一下runbroker.sh的堆內(nèi)存，如下圖所示：
若彈窗輸出下面所示的文字，則說明broker啟動(dòng)成功，自此mq就在windows環(huán)境部署成功了。我們就可以開始編碼工作了。
The broker[DESKTOP-BI4ATFQ, 192.168.237.1:10911] boot success. serializeType=JSON and name server is 127.0.0.1:98762. 訂單系統(tǒng)改造
本次的示例是關(guān)于訂單系統(tǒng)改造，用戶下單完成后，服務(wù)器需要進(jìn)行庫存扣減、訂單狀態(tài)更新、以及優(yōu)惠券、積分等邊邊角角的業(yè)務(wù)，如果順序執(zhí)行這些邏輯+網(wǎng)絡(luò)開銷，接口耗時(shí)對(duì)于用戶體驗(yàn)是非常不友好的。
所以我們?cè)趯⒎呛诵臉I(yè)務(wù)邏輯從接口串行調(diào)用中抽出，下單業(yè)務(wù)只需關(guān)注完成我們庫存扣減、訂單狀態(tài)更新就行了，剩下的業(yè)務(wù)用MQ發(fā)個(gè)消息給積分系統(tǒng)、促銷系統(tǒng)告知他們自己處理一下就行了：
首先我們引入MQ依賴腳手架：
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-spring-boot-starter</artifactId>
<version>2.1.1</version>
</dependency>同時(shí)這里我們也給出配置信息：
# mq地址端口
rocketmq.name-server=127.0.0.1:9876
# 生產(chǎn)者配置
rocketmq.producer.isOnOff=on
# 發(fā)送同一類消息設(shè)置為同一個(gè)group，保證唯一
rocketmq.producer.group=rocketmq-group
rocketmq.producer.groupName=rocketmq-group
# namesrv地址
rocketmq.producer.namesrvAddr=127.0.0.1:9876
# 設(shè)置消息最大長度 4M
rocketmq.producer.maxMessageSize=4096
# 消息發(fā)送超時(shí)時(shí)間
rocketmq.producer.sendMsgTimeout=3000
# 消息發(fā)送失敗重試次數(shù)
rocketmq.producer.retryTimesWhenSendFailed=2隨后我們?cè)O(shè)置監(jiān)聽處理關(guān)于訂單創(chuàng)建的topic消息：
@Component
@RocketMQMessageListener(consumerGroup = "${rocketmq.producer.groupName}", topic = "ORDER_ADD")
@Slf4j
public class OrderMsgListener implements RocketMQListener<Order> {
@Override
public void onMessage(Order order) {
log.info("收到訂單,訂單信息:[{}]，進(jìn)行積分系統(tǒng)、促銷系統(tǒng)、推送系統(tǒng)業(yè)務(wù)處理.....", JSONUtil.toJsonStr(order));
}
}完成后我們基于CommandLineRunner 測試一下消息發(fā)送：
@Component
@Slf4j
public class MQTest implements CommandLineRunner {
@Autowired
private RocketMQTemplate rocketMQTemplate;
@Override
public void run(String... args) throws Exception {
Order order = new Order();
order.setOrderNo("20221217001002003");
order.setUserId(1);
order.setPrice(500.00);
rocketMQTemplate.asyncSend("ORDER_ADD", MessageBuilder.withPayload(order).build(), getDefaultSendCallBack());
}
/**
* 消息處理默認(rèn)回調(diào)
* @return
*/
private SendCallback getDefaultSendCallBack() {
return new SendCallback() {
@Override
public void onSuccess(org.apache.rocketmq.client.producer.SendResult sendResult) {
log.info("MQ消息發(fā)送成功。result:{}", JSONUtil.toJsonStr(sendResult));
}
@Override
public void onException(Throwable throwable) {
log.error("MQ消息發(fā)送失敗，失敗原因:{}" + throwable.getMessage());
}
};
}
}日志如下，可以看到消息消費(fèi)成功了：
2025-02-11 10:03:14.577 INFO 14420 --- [MessageThread_1] com.sharkChili.config.OrderMsgListener : 收到訂單,訂單信息:[{"userId":1,"orderNo":"20221217001002003","price":500}]，進(jìn)行積分系統(tǒng)、促銷系統(tǒng)、推送系統(tǒng)業(yè)務(wù)處理.....
2025-02-11 10:03:14.577 INFO 14420 --- [ublicExecutor_2] com.sharkChili.runner.MQTest : MQ消息發(fā)送成功。result:{"sendStatus":"SEND_OK","msgId":"AC1E1001385418B4AAC235A7E0190000","messageQueue":{"topic":"ORDER_ADD","brokerName":"DESKTOP-DC9PSUS","queueId":2},"queueOffset":1,"offsetMsgId":"AC15733800002A9F0000000000000558","regionId":"DefaultRegion","traceOn":true}3. 如何實(shí)現(xiàn)消息過濾
設(shè)置tag消息的方式常見的是有兩種，一種是基于tag標(biāo)簽過濾，如下代碼所示，我們希望發(fā)送訂單業(yè)務(wù)即ORDER_ADD這個(gè)主題下tag標(biāo)簽為tagA的用戶收到消息，那么我們就可以通過ORDER_ADD:tagA針對(duì)topic進(jìn)行更進(jìn)一步劃分：
//創(chuàng)建訂單消息
Order order = new Order();
order.setUserId(1);
order.setOrderNo(UUID.randomUUID().toString());
order.setPrice(500);
//生成消息
Message<Order> message = MessageBuilder.withPayload(order)
.build();
//同步發(fā)送
rocketMQTemplate.syncSend("ORDER_ADD:tagA", message);對(duì)應(yīng)的監(jiān)聽者通過selectorExpression 指定標(biāo)簽即可：
@Component
@RocketMQMessageListener(consumerGroup = "${rocketmq.producer.groupName}",
topic = "ORDER_ADD",
selectorExpression = "tagA"http://訂閱tagA的消息
)
@Slf4j
public class OrderMsgListener implements RocketMQListener<Order> {
@Override
public void onMessage(Order order) {
log.info("收到訂單,訂單信息:[{}]，進(jìn)行積分系統(tǒng)、促銷系統(tǒng)、推送系統(tǒng)業(yè)務(wù)處理.....", JSONUtil.toJsonStr(order));
}
}還有一種就是基于SQL過濾，因?yàn)楸磉_(dá)式靈活，相對(duì)更強(qiáng)大一些，例如我們的消費(fèi)者只處理userId為10以內(nèi)的數(shù)據(jù)，那么消費(fèi)者的監(jiān)聽就可以按照如下姿勢(shì)進(jìn)行配置：
@Component
@RocketMQMessageListener(consumerGroup = "${rocketmq.producer.groupName}",
topic = "ORDER_ADD",
selectorType = SelectorType.SQL92,//指令類型為sql表達(dá)式
selectorExpression = "userId<10"http://過濾出id小于10的用戶的訂單
)
@Slf4j
public class OrderMsgListener implements RocketMQListener<Order> {
@Override
public void onMessage(Order order) {
log.info("收到訂單,訂單信息:[{}]，進(jìn)行積分系統(tǒng)、促銷系統(tǒng)、推送系統(tǒng)業(yè)務(wù)處理.....", JSONUtil.toJsonStr(order));
}
}發(fā)送消息時(shí)，通過headers 指定本消息條件并通過convertAndSend發(fā)送即可：
//創(chuàng)建訂單消息
Order order = new Order();
order.setUserId(1);
order.setOrderNo(UUID.randomUUID().toString());
order.setPrice(500);
//通過header攜帶條件告知當(dāng)前userId為1
Map<String, Object> headers = new HashMap<>();
headers.put("userId", 1);
//生成消息
Message<Order> message = MessageBuilder.withPayload(order)
.build();
//發(fā)送
rocketMQTemplate.convertAndSend("ORDER_ADD", message, headers);需要注意的是默認(rèn)情況下，MQ是不支持SQL表達(dá)過濾，我們需要到conf目錄下的broker.conf文件，添加enablePropertyFilter=true，然后鍵入如下指令降broker啟動(dòng)：
./mqbroker -n 127.0.0.1:9876 autoCreateTopicEnable=true -c ../conf/broker.conf4. 如何提交延時(shí)消息
延遲消息即需要消費(fèi)者過一段時(shí)間后才能消費(fèi)的消息，例如我們現(xiàn)在有個(gè)消息要求消費(fèi)者10s后才能消費(fèi)，那么我們就可以使用延遲消息，如下代碼所示：
// 創(chuàng)建延遲消息
Message<String> rocketMessage = MessageBuilder.withPayload("this is delay msg").build();
// 發(fā)送延遲消息，timeout設(shè)置為10000即10s，delayLevel表示延遲等級(jí),1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h，所以 3為10s
rocketMQTemplate.syncSend("delay_topic", rocketMessage, 10000,3);
log.info("延遲消息發(fā)送完成");消費(fèi)者代碼:
@Component
@RocketMQMessageListener(consumerGroup = "delay_msg_group", topic = "delay_topic")
@Slf4j
public class DelayMsgListener implements RocketMQListener<String> {
@Override
public void onMessage(String msg) {
log.info("收到延遲消息，消息內(nèi)容:{}", JSONUtil.toJsonStr(msg));
}
}輸出結(jié)果，可以看到確實(shí)是10s后消費(fèi)者采納看到消息并消費(fèi):
2025-02-11 10:56:58.300 INFO 18568 --- [ main] com.sharkChili.runner.MQTest : 延遲消息發(fā)送完成
2025-02-11 10:57:08.307 INFO 18568 --- [MessageThread_1] com.sharkChili.config.DelayMsgListener : 收到延遲消息，消息內(nèi)容:this is delay msg
