前置文章:
RocketMQ-基礎(chǔ)使用(一)神年,該文主要涉及MQ基礎(chǔ)、RocketMQ安裝&集群搭建拭抬、RocketMQ監(jiān)控平臺(tái)吟逝。
官方基礎(chǔ)使用樣例,很多基礎(chǔ)內(nèi)容其實(shí)官方文檔都有很詳細(xì)的說(shuō)明绒怨。日常使用如果時(shí)間充足江场,還是推薦查看官方文檔。學(xué)習(xí)官方文檔是一個(gè)良好的習(xí)慣窖逗。
零址否、本文綱要
一、RocketMQ-基礎(chǔ)使用
- 前置文章基礎(chǔ)指令
二碎紊、RocketMQ-發(fā)送消息
- 發(fā)送同步消息
- 發(fā)送異步消息
- 發(fā)送單向消息
三佑附、RocketMQ-接收消息
- 消息接收
- 消息接收-負(fù)載均衡【默認(rèn)】
- 消息接收-廣播模式
四、RocketMQ-消息類(lèi)型
- 順序消息
- 延遲消息
- 批量消息
- 過(guò)濾消息
- 事務(wù)消息
一仗考、RocketMQ-基礎(chǔ)使用
0. 前置文章基礎(chǔ)指令
Ⅰ 啟動(dòng)RocketMQ的基礎(chǔ)指令
# Start Name Server
nohup sh bin/mqnamesrv &
tail -f ~/logs/rocketmqlogs/namesrv.log
# Start Broker
nohup sh bin/mqbroker -n localhost:9876 &
tail -f ~/logs/rocketmqlogs/broker.log
指定自定義配置文件啟動(dòng)nohup sh bin/mqbroker -n localhost:9876 -c conf/broker.conf &
Ⅱ 關(guān)閉RocketMQ的基礎(chǔ)指令
# Shutdown Servers
sh bin/mqshutdown broker
sh bin/mqshutdown namesrv
二音同、RocketMQ-發(fā)送消息
發(fā)送同步消息 / 發(fā)送異步消息 / 發(fā)送單向消息
1. 發(fā)送同步消息
這種可靠性同步地發(fā)送方式使用的比較廣泛,比如:重要的消息通知秃嗜,短信通知权均。
- ① 基礎(chǔ)依賴(lài)
<dependency>
<groupId>org.apache.rocketmq</groupId>
<artifactId>rocketmq-client</artifactId>
<version>4.5.0</version>
</dependency>
- ② 同步消息代碼
/**
* 發(fā)送同步消息
*/
public class SyncProducer {
public static void main(String[] args) throws Exception {
//1.創(chuàng)建消息生產(chǎn)者producer,并制定生產(chǎn)者組名
DefaultMQProducer producer = new DefaultMQProducer("group1");
//2.指定Nameserver地址锅锨,多個(gè)NameServer則用“;”隔開(kāi)
producer.setNamesrvAddr("192.168.253.128:9876");
//3.啟動(dòng)producer
producer.start();
for (int i = 0; i < 10; i++) {
//4.創(chuàng)建消息對(duì)象叽赊,指定主題Topic、Tag和消息體
/**
* 參數(shù)一:消息主題Topic
* 參數(shù)二:消息Tag
* 參數(shù)三:消息內(nèi)容
*/
Message msg = new Message("base", "tag1", ("Hello RocketMQ [" + i + "]").getBytes());
//5.發(fā)送消息
SendResult result = producer.send(msg);
//發(fā)送狀態(tài)
SendStatus status = result.getSendStatus();
String msgId = result.getMsgId();
int queueId = result.getMessageQueue().getQueueId();
System.out.printf("發(fā)送狀態(tài):%s必搞,消息ID:%s必指,隊(duì)列:%d%n", status, msgId, queueId);
//線程睡1秒
Thread.sleep(1000);
}
//6.關(guān)閉生產(chǎn)者producer
producer.shutdown();
}
}
截取控制臺(tái)輸出
發(fā)送狀態(tài):SEND_OK,消息ID:C0A8026AC05818B4AAC28D428B270000恕洲,隊(duì)列:3
2. 發(fā)送異步消息
異步消息通常用在對(duì)響應(yīng)時(shí)間敏感的業(yè)務(wù)場(chǎng)景塔橡,即發(fā)送端不能容忍長(zhǎng)時(shí)間地等待Broker的響應(yīng)。
/**
* 發(fā)送異步消息
*/
public class AsyncProducer {
public static void main(String[] args) throws Exception {
//1.創(chuàng)建消息生產(chǎn)者producer霜第,并制定生產(chǎn)者組名
DefaultMQProducer producer = new DefaultMQProducer("group2");
//2.指定Nameserver地址葛家,多個(gè)NameServer則用“;”隔開(kāi)
producer.setNamesrvAddr("192.168.253.128:9876");
//3.啟動(dòng)producer
producer.start();
for (int i = 0; i < 10; i++) {
//4.創(chuàng)建消息對(duì)象,指定主題Topic泌类、Tag和消息體
/**
* 參數(shù)一:消息主題Topic
* 參數(shù)二:消息Tag
* 參數(shù)三:消息內(nèi)容
*/
Message msg = new Message("base", "tag2", ("AsyncMsg [" + i + "]").getBytes());
//5.發(fā)送異步消息
producer.send(msg, new SendCallback() {
/**
* 發(fā)送成功的回調(diào)函數(shù)
* @param sendResult 發(fā)送結(jié)果
*/
@Override
public void onSuccess(SendResult sendResult) {
System.out.println("發(fā)送結(jié)果:" + sendResult);
}
/**
* 發(fā)送失敗的回調(diào)函數(shù)
* @param throwable 發(fā)送異常
*/
@Override
public void onException(Throwable throwable) {
System.out.println("發(fā)送異常:" + throwable);
}
});
//線程睡1秒
Thread.sleep(1000);
}
//6.關(guān)閉生產(chǎn)者producer
producer.shutdown();
}
}
與同步消息的不同之處在于通過(guò)回調(diào)函數(shù)來(lái)獲取發(fā)送結(jié)果癞谒。
3. 發(fā)送單向消息
這種方式主要用在不特別關(guān)心發(fā)送結(jié)果的場(chǎng)景,比如:日志發(fā)送。
/**
* 發(fā)送單向消息
*/
public class OneWayProducer {
public static void main(String[] args) throws Exception, MQBrokerException {
//1.創(chuàng)建消息生產(chǎn)者producer扯俱,并制定生產(chǎn)者組名
DefaultMQProducer producer = new DefaultMQProducer("group3");
//2.指定Nameserver地址书蚪,多個(gè)NameServer則用“;”隔開(kāi)
producer.setNamesrvAddr("192.168.253.128:9876");
//3.啟動(dòng)producer
producer.start();
for (int i = 0; i < 10; i++) {
//4.創(chuàng)建消息對(duì)象,指定主題Topic迅栅、Tag和消息體
/**
* 參數(shù)一:消息主題Topic
* 參數(shù)二:消息Tag
* 參數(shù)三:消息內(nèi)容
*/
Message msg = new Message("base", "tag3", ("OneWayMsg [" + i + "]").getBytes());
//5.發(fā)送單向消息
producer.send(msg);
//線程睡1秒
Thread.sleep(1000);
}
//6.關(guān)閉生產(chǎn)者producer
producer.shutdown();
}
}
三殊校、RocketMQ-接收消息
1. 消息接收
/**
* 消息的接受者
*/
public class Consumer {
public static void main(String[] args) throws Exception {
//1.創(chuàng)建消費(fèi)者Consumer,制定消費(fèi)者組名
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group3");
//2.指定Nameserver地址
consumer.setNamesrvAddr("192.168.253.128:9876");
//3.訂閱主題Topic和Tag
consumer.subscribe("base", "tag1");
//設(shè)定消費(fèi)模式:負(fù)載均衡|廣播模式
//4.設(shè)置回調(diào)函數(shù)读存,處理消息
consumer.registerMessageListener(new MessageListenerConcurrently() {
//接收消息內(nèi)容
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list,
ConsumeConcurrentlyContext consumeConcurrentlyContext) {
System.out.println(list);
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
//5.啟動(dòng)消費(fèi)者consumer
consumer.start();
}
}
2. 消息接收-負(fù)載均衡【默認(rèn)】
消費(fèi)者采用負(fù)載均衡方式消費(fèi)消息为流,多個(gè)消費(fèi)者共同消費(fèi)隊(duì)列消息,每個(gè)消費(fèi)者處理的消息不同让簿【床欤【默認(rèn)的消息消費(fèi)方式】
consumer.setMessageModel(MessageModel.CLUSTERING);
3. 消息接收-廣播模式
消費(fèi)者采用廣播的方式消費(fèi)消息,每個(gè)消費(fèi)者消費(fèi)的消息都是相同的尔当。
consumer.setMessageModel(MessageModel.BROADCASTING);
四莲祸、RocketMQ-消息類(lèi)型
1. 順序消息
- ① 基礎(chǔ)分析
假定一個(gè)訂單的順序流程是:創(chuàng)建、付款椭迎、推送锐帜、完成。有張三畜号、李四兩人進(jìn)行訂單業(yè)務(wù)缴阎。
a、全局有序:
張三所有消息消費(fèi)完简软,再消費(fèi)李四消息蛮拔,且內(nèi)部有序;
一個(gè)Borker痹升,一個(gè)MessageQueue建炫;
b、局部有序:
只要保證各自消息內(nèi)部的有序消費(fèi)视卢,交替消費(fèi)兩者的消息是可以的踱卵;
一個(gè)Borker,多個(gè)MessageQueue据过,一個(gè)MessageQueue對(duì)應(yīng)一個(gè)訂單。
所以妒挎,一般僅需保證局部有序即可绳锅。
實(shí)現(xiàn)方式:同一個(gè)用戶的一個(gè)業(yè)務(wù)消息放到同一個(gè)隊(duì)列,比如:訂單號(hào)相同的消息進(jìn)同一個(gè)隊(duì)列酝掩。
- ② 代碼實(shí)現(xiàn)
Ⅰ 消息生產(chǎn)者producer的核心代碼
/**
* 參數(shù)一: 消息對(duì)象
* 參數(shù)二: 消息隊(duì)列選擇器 MessageQueueSelector
* 參數(shù)三: 選擇隊(duì)列業(yè)務(wù)標(biāo)識(shí)鳞芙,此處為訂單ID
*/
SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
/**
*
* @param list 消息隊(duì)列
* @param message 消息對(duì)象
* @param o 業(yè)務(wù)標(biāo)識(shí)的參數(shù)
* @return 消息隊(duì)列
*/
@Override
public MessageQueue select(List<MessageQueue> list, Message message, Object o) {
long orderId = (long) o;
long index = orderId % list.size(); //訂單ID一致,則取模結(jié)果一致,最終選擇的隊(duì)列一致
return list.get((int) index);
}
}, order.getOrderId());
Ⅱ 消息消費(fèi)者consumer的核心代碼
此處是通過(guò)有序消息監(jiān)聽(tīng)MessageListenerOrderly來(lái)實(shí)現(xiàn)的
//4.注冊(cè)消息監(jiān)聽(tīng)器
consumer.registerMessageListener(new MessageListenerOrderly() {
@Override
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {
for (MessageExt messageExt : list) {
System.out.println("線程名稱(chēng):" + Thread.currentThread().getName() + " → " +
"消費(fèi)消息:" + new String(messageExt.getBody()));
}
return ConsumeOrderlyStatus.SUCCESS;
}
});
2. 延遲消息
使用場(chǎng)景:比如電商里原朝,提交了一個(gè)訂單就可以發(fā)送一個(gè)延時(shí)消息驯嘱,1h后去檢查這個(gè)訂單的狀態(tài),如果還是未付款就取消訂單釋放庫(kù)存喳坠。
延遲消息使用限制:
// org/apache/rocketmq/store/config/MessageStoreConfig.java
private String messageDelayLevel = "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
Ⅰ 消息生產(chǎn)者producer的核心代碼
msg.setDelayTimeLevel(2);
Ⅱ 消息消費(fèi)者consumer的核心代碼
無(wú)需調(diào)整鞠评。
注意:受限于網(wǎng)絡(luò)情況,實(shí)際的延遲往往大于設(shè)置的延遲壕鹉。
3. 批量消息
批量發(fā)送消息能顯著提高傳遞小消息的性能剃幌。
限制:
a、相同的topic晾浴;
b负乡、相同的waitStoreMsgOK;
c脊凰、不能是延時(shí)消息抖棘;
d、總大小不應(yīng)超過(guò)4MB狸涌。
Ⅰ 消息生產(chǎn)者producer的核心代碼
List<Message> messageList = new ArrayList<>();
Ⅱ 消息消費(fèi)者consumer的核心代碼
無(wú)需調(diào)整钉答。
4. 過(guò)濾消息
一般過(guò)濾消息可通過(guò) TAG / SQL92標(biāo)準(zhǔn) 來(lái)進(jìn)行過(guò)濾
Ⅰ 消息生產(chǎn)者producer的核心代碼
//方式一:通過(guò)Tag過(guò)濾的使用方法,消息發(fā)送方不做調(diào)整
//...
//方式二:通過(guò)sql過(guò)濾的使用方法杈抢,使用putUserProperty設(shè)置一些消息屬性
msg.putUserProperty("a", String.valueOf(i));
Ⅱ 消息消費(fèi)者consumer的核心代碼
//方式一:通過(guò)Tag過(guò)濾的使用方法数尿,consumer使用" || "分隔訂閱不同的Tag即可
consumer.subscribe("TOPIC", "TAGA || TAGB || TAGC");
//方式二:通過(guò)sql過(guò)濾的使用方法,通過(guò)MessageSelector消息選擇器的bySql方法過(guò)濾消息
consumer.subscribe("topic_sql_filter", MessageSelector.bySql("num > 5"));
方式二如果報(bào)錯(cuò):The broker does not support consumer to filter message by SQL92
惶楼,
則需要在我們對(duì)應(yīng)的Broker配置文件內(nèi)做調(diào)整右蹦,添加enablePropertyFilter=true
,重啟服務(wù)即可生效歼捐。
關(guān)于SQL92基礎(chǔ)語(yǔ)法何陆,RocketMQ只定義了一些基本語(yǔ)法來(lái)支持這個(gè)特性:
數(shù)值比較,比如:>豹储,>=贷盲,<,<=剥扣,BETWEEN巩剖,=;
字符比較钠怯,比如:=佳魔,<>,IN晦炊;
IS NULL 或者 IS NOT NULL鞠鲜;
邏輯符號(hào) AND宁脊,OR,NOT贤姆;
常量支持類(lèi)型為:
數(shù)值榆苞,比如:123,3.1415霞捡;
字符坐漏,比如:'abc',必須用單引號(hào)包裹起來(lái)弄砍;
NULL仙畦,特殊的常量
布爾值,TRUE 或 FALSE
注意:只有使用push模式的消費(fèi)者才能用使用SQL92標(biāo)準(zhǔn)的sql語(yǔ)句音婶。
5. 事務(wù)消息
在【分布式事務(wù)-可靠消息最終一致性】的解決方案內(nèi)使用的就是事務(wù)消息慨畸。
事務(wù)消息流程:正常事務(wù)消息的發(fā)送及提交,以及事務(wù)消息的補(bǔ)償【事務(wù)狀態(tài)回查】衣式;
事務(wù)狀態(tài):
LocalTransactionState.COMMIT_MESSAGE 提交狀態(tài) 允許消費(fèi)消息寸士;
LocalTransactionState.ROLLBACK_MESSAGE 回滾狀態(tài) 刪除消息,不允許被消費(fèi)碴卧;
LocalTransactionState.UNKNOW 中間狀態(tài) 需要回查事務(wù)弱卡。
Ⅰ 消息生產(chǎn)者producer的核心代碼
/**
* 發(fā)送同步消息
*/
public class Producer {
public static void main(String[] args) throws Exception {
//1.創(chuàng)建消息生產(chǎn)者producer,并制定生產(chǎn)者組名
TransactionMQProducer producer = new TransactionMQProducer("group1");
//2.指定Nameserver地址住册,多個(gè)NameServer則用“;”隔開(kāi)
producer.setNamesrvAddr("192.168.253.128:9876");
//3.設(shè)置消息事務(wù)的監(jiān)聽(tīng)器
producer.setTransactionListener(new TransactionListener() {
/**
* 在該方法中執(zhí)行本地的事務(wù)
* @param message 消息
* @param o
* @return 事務(wù)狀態(tài)
*/
@Override
public LocalTransactionState executeLocalTransaction(Message message, Object o) {
String messageTags = message.getTags();
if (StringUtils.equals("TagA", messageTags)) {
return LocalTransactionState.COMMIT_MESSAGE;
} else if (StringUtils.equals("TagB", messageTags)) {
return LocalTransactionState.ROLLBACK_MESSAGE;
} else {
return LocalTransactionState.UNKNOW;
}
}
/**
* 該方法進(jìn)行MQ事務(wù)狀態(tài)的回查
* @param messageExt 消息
* @return 事務(wù)狀態(tài)
*/
@Override
public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
System.out.println("消息的Tag:" + messageExt.getTags());
return LocalTransactionState.COMMIT_MESSAGE;
}
});
String[] tags = new String[]{"TagA", "TagB", "TagC"};
//4.啟動(dòng)producer
producer.start();
for (int i = 0; i < 3; i++) {
//5.創(chuàng)建消息對(duì)象婶博,指定Topic、Tag荧飞、消息體
Message msg = new Message("topic_transaction", tags[i],
(tags[i] + " Hello transactionMsg " + i).getBytes(StandardCharsets.UTF_8));
//6.發(fā)送消息
SendResult sendResult = producer.sendMessageInTransaction(msg, null);
System.out.println("發(fā)送結(jié)果:" + sendResult);
Thread.sleep(1000);
}
//7.關(guān)閉生產(chǎn)者producer凡人,此處需要回查,所以不關(guān)閉
//producer.shutdown();
}
}
Ⅱ 消息消費(fèi)者consumer的核心代碼
/**
* 消息的接受者
*/
public class Consumer {
public static void main(String[] args) throws Exception {
//1.創(chuàng)建消費(fèi)者Consumer叹阔,制定消費(fèi)者組名
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("group1");
//2.指定Nameserver地址
consumer.setNamesrvAddr("192.168.253.128:9876");
//3.訂閱主題Topic和Tag
consumer.subscribe("topic_transaction", "*");
//4.注冊(cè)消息監(jiān)聽(tīng)器
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list,
ConsumeConcurrentlyContext consumeConcurrentlyContext) {
for (MessageExt messageExt : list) {
System.out.println("消費(fèi)消息:" + new String(messageExt.getBody()));
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
//5.啟動(dòng)消費(fèi)者consumer
consumer.start();
System.out.println("消費(fèi)者啟動(dòng)了...");
}
}
注意:事務(wù)消息不支持延時(shí)消息和批量消息挠轴。
五、結(jié)尾
以上即為RocketMQ-基礎(chǔ)使用(二)的全部?jī)?nèi)容耳幢,感謝閱讀岸晦。