溫馨提示×
您好，登錄后才能下訂單哦！
點(diǎn)擊 登錄注冊(cè) 即表示同意《億速云用戶服務(wù)條款》
您好，登錄后才能下訂單哦！
這篇文章給大家分享的是有關(guān)RocketMQ如何安裝使用的內(nèi)容。小編覺得挺實(shí)用的，因此分享給大家做個(gè)參考，一起跟隨小編過來看看吧。
一、服務(wù)端安裝部署
我是在虛擬機(jī)中的CentOS6.5中進(jìn)行部署。
1.下載程序
2.tar -xvf alibaba-rocketmq-3.0.7.tar.gz 解壓到適當(dāng)?shù)哪夸浫?opt/目錄
3.啟動(dòng)RocketMQ：進(jìn)入rocketmq/bin 目錄 執(zhí)行
nohup sh mqnamesrv & |
4.啟動(dòng)Broker，設(shè)置對(duì)應(yīng)的NameServer
nohup sh mqbroker -n "127.0.0.1:9876" & |
二、編寫客戶端
可以查看sameple中的quickstart源碼 1.Consumer 消息消費(fèi)者
/**
* Consumer，訂閱消息
*/
public class Consumer {
public static void main(String[] args) throws InterruptedException, MQClientException {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("QuickStartConsumer");
consumer.setNamesrvAddr("127.0.0.1:9876");
consumer.setInstanceName("QuickStartConsumer");
consumer.subscribe("QuickStart", "*");
consumer.registerMessageListener(new MessageListenerConcurrently() {
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
ConsumeConcurrentlyContext context) {
System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs);
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
});
consumer.start();
System.out.println("Consumer Started.");
}
} |
2.Producer消息生產(chǎn)者
/**
* Producer，發(fā)送消息
*
*/
public class Producer {
public static void main(String[] args) throws MQClientException, InterruptedException {
DefaultMQProducer producer = new DefaultMQProducer("QuickStartProducer");
producer.setNamesrvAddr("127.0.0.1:9876");
producer.setInstanceName("QuickStartProducer");
producer.start();
for (int i = 0; i < 1000; i++) {
try {
Message msg = new Message("QuickStart",// topic
"TagA",// tag
("Hello RocketMQ ,QuickStart" + i).getBytes()// body
);
SendResult sendResult = producer.send(msg);
System.out.println(sendResult);
}
catch (Exception e) {
e.printStackTrace();
Thread.sleep(1000);
}
}
producer.shutdown();
}
} |
3.首先運(yùn)行Consumer程序，一直在運(yùn)行狀態(tài)接收服務(wù)器端推送過來的消息
23:18:07.587 [main] DEBUG i.n.c.MultithreadEventLoopGroup - -Dio.netty.eventLoopThreads: 16 23:18:07.591 [main] DEBUG i.n.util.internal.PlatformDependent - Platform: Windows 23:18:07.592 [main] DEBUG i.n.util.internal.PlatformDependent - Java version: 7 23:18:07.592 [main] DEBUG i.n.util.internal.PlatformDependent - -Dio.netty.noUnsafe: false 23:18:07.593 [main] DEBUG i.n.util.internal.PlatformDependent0 - java.nio.ByteBuffer.cleaner: available 23:18:07.593 [main] DEBUG i.n.util.internal.PlatformDependent0 - java.nio.Buffer.address: available 23:18:07.593 [main] DEBUG i.n.util.internal.PlatformDependent0 - sun.misc.Unsafe.theUnsafe: available 23:18:07.593 [main] DEBUG i.n.util.internal.PlatformDependent0 - sun.misc.Unsafe.copyMemory: available 23:18:07.593 [main] DEBUG i.n.util.internal.PlatformDependent0 - java.nio.Bits.unaligned: true 23:18:07.594 [main] DEBUG i.n.util.internal.PlatformDependent - sun.misc.Unsafe: available 23:18:07.594 [main] DEBUG i.n.util.internal.PlatformDependent - -Dio.netty.noJavassist: false 23:18:07.594 [main] DEBUG i.n.util.internal.PlatformDependent - Javassist: unavailable 23:18:07.594 [main] DEBUG i.n.util.internal.PlatformDependent - You don't have Javassist in your class path or you don't have enough permission to load dynamically generated classes. Please check the configuration for better performance. 23:18:07.595 [main] DEBUG i.n.util.internal.PlatformDependent - -Dio.netty.noPreferDirect: false 23:18:07.611 [main] DEBUG io.netty.channel.nio.NioEventLoop - -Dio.netty.noKeySetOptimization: false 23:18:07.611 [main] DEBUG io.netty.channel.nio.NioEventLoop - -Dio.netty.selectorAutoRebuildThreshold: 512 23:18:08.355 [main] DEBUG i.n.util.internal.ThreadLocalRandom - -Dio.netty.initialSeedUniquifier: 0x8c0d4793e5820c31 23:18:08.446 [NettyClientWorkerThread_1] DEBUG io.netty.util.ResourceLeakDetector - -Dio.netty.noResourceLeakDetection: false Consumer Started. |
4.再次運(yùn)行Producer程序，生成消息并發(fā)送到Broker，Producer的日志沖沒了，但是可以看到Broker推送到Consumer的一條消息
ConsumeMessageThread-QuickStartConsumer-3 Receive New Messages: [MessageExt [queueId=0, storeSize=150,
queueOffset=244, sysFlag=0, bornTimestamp=1400772029972, bornHost=/10.162.0.7:54234,
storeTimestamp=1400772016017, storeHost=/127.0.0.1:10911, msgId=0A0A0A5900002A9F0000000000063257,
commitLogOffset=406103, bodyCRC=112549959, reconsumeTimes=0, preparedTransactionOffset=0,
toString()=Message [topic=QuickStart, flag=0,
properties={TAGS=TagA, WAIT=true, MAX_OFFSET=245, MIN_OFFSET=0}, body=29]]] |
三、Consumer最佳實(shí)踐
1.消費(fèi)過程要做到冪等（即消費(fèi)端去重）
RocketMQ無法做到消息重復(fù)，所以如果業(yè)務(wù)對(duì)消息重復(fù)非常敏感，務(wù)必要在業(yè)務(wù)層面去重，有以下一些方式：
（1）.將消息的唯一鍵，可以是MsgId，也可以是消息內(nèi)容中的唯一標(biāo)識(shí)字段，例如訂單ID，消費(fèi)之前判斷是否在DB或Tair（全局KV存儲(chǔ)）中存在，如果不存在則插入，并消費(fèi)，否則跳過。（實(shí)踐過程要考慮原子性問題，判斷是否存在可以嘗試插入，如果報(bào)主鍵沖突，則插入失敗，直接跳過） msgid一定是全局唯一的標(biāo)識(shí)符，但是可能會(huì)存在同樣的消息有兩個(gè)不同的msgid的情況（有多種原因），這種情況可能會(huì)使業(yè)務(wù)上重復(fù)，建議最好使用消息體中的唯一標(biāo)識(shí)字段去重
（2）.使業(yè)務(wù)層面的狀態(tài)機(jī)去重
2.批量方式消費(fèi)
如果業(yè)務(wù)流程支持批量方式消費(fèi)，則可以很大程度上的提高吞吐量，可以通過設(shè)置Consumer的consumerMessageBatchMaxSize參數(shù)，默認(rèn)是1，即一次消費(fèi)一條參數(shù)
3.跳過非重要的消息
發(fā)生消息堆積時(shí)，如果消費(fèi)速度一直跟不上發(fā)送速度，可以選擇丟棄不重要的消息
@Override
public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
ConsumeConcurrentlyContext context) {
System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs);
long offset=msgs.get(0).getQueueOffset();
String maxOffset=msgs.get(0).getProperty(MessageConst.PROPERTY_MAX_OFFSET);
long diff=Long.parseLong(maxOffset)-offset;
if(diff>100000){
//處理消息堆積情況
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
}
return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
} |
如以上代碼所示，當(dāng)某個(gè)隊(duì)列的消息數(shù)堆積到 100000 條以上，則嘗試丟棄部分或全部消息，這樣就可以快速追上發(fā)送消息的速度
4.優(yōu)化沒條消息消費(fèi)過程
舉例如下，某條消息的消費(fèi)過程如下
1. 根據(jù)消息從 DB 查詢數(shù)據(jù) 1
2. 根據(jù)消息從 DB 查詢數(shù)據(jù)2
3. 復(fù)雜的業(yè)務(wù)計(jì)算
4. 向 DB 插入數(shù)據(jù)3
5. 向 DB 插入數(shù)據(jù) 4
這條消息的消費(fèi)過程與 DB 交互了 4 次，如果按照每次 5ms 計(jì)算，那么總共耗時(shí) 20ms，假設(shè)業(yè)務(wù)計(jì)算耗時(shí) 5ms，那么總過耗時(shí) 25ms，如果能把 4 次 DB 交互優(yōu)化為 2 次，那么總耗時(shí)就可以優(yōu)化到 15ms，也就是說總體性能提高了 40%。
對(duì)于 Mysql 等 DB，如果部署在磁盤，那么與 DB 進(jìn)行交互，如果數(shù)據(jù)沒有命中 cache，每次交互的 RT 會(huì)直線上升， 如果采用 SSD，則 RT 上升趨勢(shì)要明顯好于磁盤。
個(gè)別應(yīng)用可能會(huì)遇到這種情況：在線下壓測(cè)消費(fèi)過程中，db 表現(xiàn)非常好，每次 RT 都很短，但是上線運(yùn)行一段時(shí)間，RT 就會(huì)變長，消費(fèi)吞吐量直線下降
主要原因是線下壓測(cè)時(shí)間過短，線上運(yùn)行一段時(shí)間后，cache 命中率下降，那么 RT 就會(huì)增加。建議在線下壓測(cè)時(shí)，要測(cè)試足夠長時(shí)間，盡可能模擬線上環(huán)境，壓測(cè)過程中，數(shù)據(jù)的分布也很重要，數(shù)據(jù)不同，可能 cache 的命中率也會(huì)完全不同
四、Producer最佳實(shí)踐
1.發(fā)送消息注意事項(xiàng)
（1） 一個(gè)應(yīng)用盡可能用一個(gè) Topic，消息子類型用 tags 來標(biāo)識(shí)，tags 可以由應(yīng)用自由設(shè)置。只有發(fā)送消息設(shè)置了tags，消費(fèi)方在訂閱消息時(shí)，才可以利用 tags 在 broker 做消息過濾。
（2）每個(gè)消息在業(yè)務(wù)層面的唯一標(biāo)識(shí)碼，要設(shè)置到 keys 字段，方便將來定位消息丟失問題。服務(wù)器會(huì)為每個(gè)消息創(chuàng)建索引（哈希索引），應(yīng)用可以通過 topic，key 來查詢這條消息內(nèi)容，以及消息被誰消費(fèi)。由于是哈希索引，請(qǐng)務(wù)必保證 key 盡可能唯一，這樣可以避免潛在的哈希沖突。
（3）消息發(fā)送成功或者失敗，要打印消息日志，務(wù)必要打印 sendresult 和 key 字段
（4）send 消息方法，只要不拋異常，就代表發(fā)送成功。但是發(fā)送成功會(huì)有多個(gè)狀態(tài)，在 sendResult 里定義
SEND_OK：消息發(fā)送成功
FLUSH_DISK_TIMEOUT：消息發(fā)送成功，但是服務(wù)器刷盤超時(shí)，消息已經(jīng)進(jìn)入服務(wù)器隊(duì)列，只有此時(shí)服務(wù)器宕機(jī)，消息才會(huì)丟失
FLUSH_SLAVE_TIMEOUT：消息發(fā)送成功，但是服務(wù)器同步到 Slave 時(shí)超時(shí)，消息已經(jīng)進(jìn)入服務(wù)器隊(duì)列，只有此時(shí)服務(wù)器宕機(jī)，消息才會(huì)丟失
SLAVE_NOT_AVAILABLE：消息發(fā)送成功，但是此時(shí) slave 不可用，消息已經(jīng)進(jìn)入服務(wù)器隊(duì)列，只有此時(shí)服務(wù)器宕機(jī)，消息才會(huì)丟失。對(duì)于精確發(fā)送順序消息的應(yīng)用，由于順序消息的局限性，可能會(huì)涉及到主備自動(dòng)切換問題，所以如果sendresult 中的 status 字段不等于 SEND_OK，就應(yīng)該嘗試重試。對(duì)于其他應(yīng)用，則沒有必要這樣
（5）對(duì)于消息不可丟失應(yīng)用，務(wù)必要有消息重發(fā)機(jī)制
2.消息發(fā)送失敗處理
Producer 的 send 方法本身支持內(nèi)部重試，重試邏輯如下：
（1） 至多重試 3 次
（2） 如果發(fā)送失敗，則輪轉(zhuǎn)到下一個(gè) Broker
（3） 這個(gè)方法的總耗時(shí)時(shí)間不超過 sendMsgTimeout 設(shè)置的值，默認(rèn) 10s所以，如果本身向 broker 發(fā)送消息產(chǎn)生超時(shí)異常，就不會(huì)再做重試
如果調(diào)用 send 同步方法發(fā)送失敗，則嘗試將消息存儲(chǔ)到 db，由后臺(tái)線程定時(shí)重試，保證消息一定到達(dá) Broker。
上述 db 重試方式為什么沒有集成到 MQ 客戶端內(nèi)部做，而是要求應(yīng)用自己去完成，基于以下幾點(diǎn)考慮：
（1）MQ 的客戶端設(shè)計(jì)為無狀態(tài)模式，方便任意的水平擴(kuò)展，且對(duì)機(jī)器資源的消耗僅僅是 cpu、內(nèi)存、網(wǎng)絡(luò)
（2）如果 MQ 客戶端內(nèi)部集成一個(gè) KV 存儲(chǔ)模塊，那么數(shù)據(jù)只有同步落盤才能較可靠，而同步落盤本身性能開銷較大，所以通常會(huì)采用異步落盤，又由于應(yīng)用關(guān)閉過程不受 MQ 運(yùn)維人員控制，可能經(jīng)常會(huì)發(fā)生 kill -9 這樣暴力方式關(guān)閉，造成數(shù)據(jù)沒有及時(shí)落盤而丟失
（3）Producer 所在機(jī)器的可靠性較低，一般為虛擬機(jī)，不適合存儲(chǔ)重要數(shù)據(jù)。 綜上，建議重試過程交由應(yīng)用來控制。
3.選擇 oneway 形式發(fā)送
一個(gè) RPC 調(diào)用，通常是這樣一個(gè)過程
（1）客戶端發(fā)送請(qǐng)求到服務(wù)器
（2）服務(wù)器處理該請(qǐng)求
（3）服務(wù)器向客戶端返回應(yīng)答
所以一個(gè) RPC 的耗時(shí)時(shí)間是上述三個(gè)步驟的總和，而某些場(chǎng)景要求耗時(shí)非常短，但是對(duì)可靠性要求并不高，例如日志收集類應(yīng)用，此類應(yīng)用可以采用 oneway 形式調(diào)用，oneway 形式只發(fā)送請(qǐng)求不等待應(yīng)答，而發(fā)送請(qǐng)求在客戶端實(shí)現(xiàn)層面僅僅是一個(gè) os 系統(tǒng)調(diào)用的開銷，即將數(shù)據(jù)寫入客戶端的 socket 緩沖區(qū)，此過程耗時(shí)通常在微秒級(jí)。
RocketMQ不止可以直接推送消息，在消費(fèi)端注冊(cè)監(jiān)聽器進(jìn)行監(jiān)聽，還可以由消費(fèi)端決定自己去拉取數(shù)據(jù)
/**
* PullConsumer，訂閱消息
*/
public class PullConsumer {
//Java緩存
private static final Map<MessageQueue, Long> offseTable = new HashMap<MessageQueue, Long>();
public static void main(String[] args) throws MQClientException {
DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("PullConsumerGroup");
consumer.setNamesrvAddr("127.0.0.1:9876");
consumer.start();
//拉取訂閱主題的隊(duì)列，默認(rèn)隊(duì)列大小是4
Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues("TopicTestMapBody");
for (MessageQueue mq : mqs) {
System.out.println("Consume from the queue: " + mq);
SINGLE_MQ:while(true){
try {
PullResult pullResult =
consumer.pullBlockIfNotFound(mq, null, getMessageQueueOffset(mq), 32);
List<MessageExt> list=pullResult.getMsgFoundList();
if(list!=null&&list.size()<100){
for(MessageExt msg:list){
System.out.println(SerializableInterface.deserialize(msg.getBody()));
}
}
System.out.println(pullResult.getNextBeginOffset());
putMessageQueueOffset(mq, pullResult.getNextBeginOffset());
switch (pullResult.getPullStatus()) {
case FOUND:
// TODO
break;
case NO_MATCHED_MSG:
break;
case NO_NEW_MSG:
break SINGLE_MQ;
case OFFSET_ILLEGAL:
break;
default:
break;
}
}
catch (Exception e) {
e.printStackTrace();
}
}
}
consumer.shutdown();
}
private static void putMessageQueueOffset(MessageQueue mq, long offset) {
offseTable.put(mq, offset);
}
private static long getMessageQueueOffset(MessageQueue mq) {
Long offset = offseTable.get(mq);
if (offset != null){
System.out.println(offset);
return offset;
}
return 0;
} |
剛開始的沒有細(xì)看PullResult對(duì)象，以為拉取到的結(jié)果沒有MessageExt對(duì)象還跑到群里面問別人，犯2了
特別要注意 靜態(tài)變量offsetTable的作用，拉取的是按照從offset（理解為下標(biāo)）位置開始拉取，拉取N條，offsetTable記錄下次拉取的offset位置。
感謝各位的閱讀！關(guān)于“RocketMQ如何安裝使用”這篇文章就分享到這里了，希望以上內(nèi)容可以對(duì)大家有一定的幫助，讓大家可以學(xué)到更多知識(shí)，如果覺得文章不錯(cuò)，可以把它分享出去讓更多的人看到吧！
免責(zé)聲明：本站發(fā)布的內(nèi)容（圖片、視頻和文字）以原創(chuàng)、轉(zhuǎn)載和分享為主，文章觀點(diǎn)不代表本網(wǎng)站立場(chǎng)，如果涉及侵權(quán)請(qǐng)聯(lián)系站長郵箱：is@yisu.com進(jìn)行舉報(bào)，并提供相關(guān)證據(jù)，一經(jīng)查實(shí)，將立刻刪除涉嫌侵權(quán)內(nèi)容。
億速云公眾號(hào)
手機(jī)網(wǎng)站二維碼
