RocketMQ如何安裝使用

189次閱讀

沒有評論

共計 8814 個字符，預計需要花費 23 分鐘才能閱讀完成。

這篇文章給大家分享的是有關 RocketMQ 如何安裝使用的內容。丸趣 TV 小編覺得挺實用的，因此分享給大家做個參考，一起跟隨丸趣 TV 小編過來看看吧。

一、服務端安裝部署

我是在虛擬機中的 CentOS6.5 中進行部署。

1. 下載程序

2.tar -xvf alibaba-rocketmq-3.0.7.tar.gz 解壓到適當的目錄如 /opt/ 目錄

3. 啟動 RocketMQ：進入 rocketmq/bin 目錄執行

nohup sh mqnamesrv

4. 啟動 Broker，設置對應的 NameServer

nohup sh mqbroker -n  127.0.0.1:9876

二、編寫客戶端

可以查看 sameple 中的 quickstart 源碼 1.Consumer 消息消費者

/**
 * 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 消息生產者

/**
 * Producer，發送消息
 *
 */
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. 首先運行 Consumer 程序，一直在運行狀態接收服務器端推送過來的消息

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. 再次運行 Producer 程序，生成消息并發送到 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 最佳實踐

1. 消費過程要做到冪等（即消費端去重）

RocketMQ 無法做到消息重復，所以如果業務對消息重復非常敏感，務必要在業務層面去重，有以下一些方式：

（1）. 將消息的唯一鍵，可以是 MsgId，也可以是消息內容中的唯一標識字段，例如訂單 ID，消費之前判斷是否在 DB 或 Tair（全局 KV 存儲）中存在，如果不存在則插入，并消費，否則跳過。（實踐過程要考慮原子性問題，判斷是否存在可以嘗試插入，如果報主鍵沖突，則插入失敗，直接跳過）msgid 一定是全局唯一的標識符，但是可能會存在同樣的消息有兩個不同的 msgid 的情況（有多種原因），這種情況可能會使業務上重復，建議最好使用消息體中的唯一標識字段去重

（2）. 使業務層面的狀態機去重

2. 批量方式消費

如果業務流程支持批量方式消費，則可以很大程度上的提高吞吐量，可以通過設置 Consumer 的 consumerMessageBatchMaxSize 參數，默認是 1，即一次消費一條參數

3. 跳過非重要的消息

發生消息堆積時，如果消費速度一直跟不上發送速度，可以選擇丟棄不重要的消息

@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;
 }

如以上代碼所示，當某個隊列的消息數堆積到 100000 條以上，則嘗試丟棄部分或全部消息，這樣就可以快速追上發送消息的速度

4. 優化沒條消息消費過程

舉例如下，某條消息的消費過程如下

1. 根據消息從 DB 查詢數據 1

2. 根據消息從 DB 查詢數據 2

3. 復雜的業務計算

4. 向 DB 插入數據 3

5. 向 DB 插入數據 4

這條消息的消費過程與 DB 交互了 4 次，如果按照每次 5ms 計算，那么總共耗時 20ms，假設業務計算耗時 5ms，那么總過耗時 25ms，如果能把 4 次 DB 交互優化為 2 次，那么總耗時就可以優化到 15ms，也就是說總體性能提高了 40%。

對于 Mysql 等 DB，如果部署在磁盤，那么與 DB 進行交互，如果數據沒有命中 cache，每次交互的 RT 會直線上升，如果采用 SSD，則 RT 上升趨勢要明顯好于磁盤。

個別應用可能會遇到這種情況：在線下壓測消費過程中，db 表現非常好，每次 RT 都很短，但是上線運行一段時間，RT 就會變長，消費吞吐量直線下降

主要原因是線下壓測時間過短，線上運行一段時間后，cache 命中率下降，那么 RT 就會增加。建議在線下壓測時，要測試足夠長時間，盡可能模擬線上環境，壓測過程中，數據的分布也很重要，數據不同，可能 cache 的命中率也會完全不同

四、Producer 最佳實踐

1. 發送消息注意事項

（1）一個應用盡可能用一個 Topic，消息子類型用 tags 來標識，tags 可以由應用自由設置。只有發送消息設置了 tags，消費方在訂閱消息時，才可以利用 tags 在 broker 做消息過濾。

（2）每個消息在業務層面的唯一標識碼，要設置到 keys 字段，方便將來定位消息丟失問題。服務器會為每個消息創建索引（哈希索引），應用可以通過 topic，key 來查詢這條消息內容，以及消息被誰消費。由于是哈希索引，請務必保證 key 盡可能唯一，這樣可以避免潛在的哈希沖突。

（3）消息發送成功或者失敗，要打印消息日志，務必要打印 sendresult 和 key 字段

（4）send 消息方法，只要不拋異常，就代表發送成功。但是發送成功會有多個狀態，在 sendResult 里定義

SEND_OK：消息發送成功

FLUSH_DISK_TIMEOUT：消息發送成功，但是服務器刷盤超時，消息已經進入服務器隊列，只有此時服務器宕機，消息才會丟失

FLUSH_SLAVE_TIMEOUT：消息發送成功，但是服務器同步到 Slave 時超時，消息已經進入服務器隊列，只有此時服務器宕機，消息才會丟失

SLAVE_NOT_AVAILABLE：消息發送成功，但是此時 slave 不可用，消息已經進入服務器隊列，只有此時服務器宕機，消息才會丟失。對于精確發送順序消息的應用，由于順序消息的局限性，可能會涉及到主備自動切換問題，所以如果 sendresult 中的 status 字段不等于 SEND_OK，就應該嘗試重試。對于其他應用，則沒有必要這樣

（5）對于消息不可丟失應用，務必要有消息重發機制

2. 消息發送失敗處理

Producer 的 send 方法本身支持內部重試，重試邏輯如下：

（1）至多重試 3 次

（2）如果發送失敗，則輪轉到下一個 Broker

（3）這個方法的總耗時時間不超過 sendMsgTimeout 設置的值，默認 10s 所以，如果本身向 broker 發送消息產生超時異常，就不會再做重試

如果調用 send 同步方法發送失敗，則嘗試將消息存儲到 db，由后臺線程定時重試，保證消息一定到達 Broker。

上述 db 重試方式為什么沒有集成到 MQ 客戶端內部做，而是要求應用自己去完成，基于以下幾點考慮：

（1）MQ 的客戶端設計為無狀態模式，方便任意的水平擴展，且對機器資源的消耗僅僅是 cpu、內存、網絡

（2）如果 MQ 客戶端內部集成一個 KV 存儲模塊，那么數據只有同步落盤才能較可靠，而同步落盤本身性能開銷較大，所以通常會采用異步落盤，又由于應用關閉過程不受 MQ 運維人員控制，可能經常會發生 kill -9 這樣暴力方式關閉，造成數據沒有及時落盤而丟失

（3）Producer 所在機器的可靠性較低，一般為虛擬機，不適合存儲重要數據。綜上，建議重試過程交由應用來控制。

3. 選擇 oneway 形式發送

一個 RPC 調用，通常是這樣一個過程

（1）客戶端發送請求到服務器

（2）服務器處理該請求

（3）服務器向客戶端返回應答

所以一個 RPC 的耗時時間是上述三個步驟的總和，而某些場景要求耗時非常短，但是對可靠性要求并不高，例如日志收集類應用，此類應用可以采用 oneway 形式調用，oneway 形式只發送請求不等待應答，而發送請求在客戶端實現層面僅僅是一個 os 系統調用的開銷，即將數據寫入客戶端的 socket 緩沖區，此過程耗時通常在微秒級。

RocketMQ 不止可以直接推送消息，在消費端注冊監聽器進行監聽，還可以由消費端決定自己去拉取數據

/**
 * 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();
 // 拉取訂閱主題的隊列，默認隊列大小是 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;
 }

剛開始的沒有細看 PullResult 對象，以為拉取到的結果沒有 MessageExt 對象還跑到群里面問別人，犯 2 了

特別要注意靜態變量 offsetTable 的作用，拉取的是按照從 offset（理解為下標）位置開始拉取，拉取 N 條，offsetTable 記錄下次拉取的 offset 位置。

感謝各位的閱讀！關于“RocketMQ 如何安裝使用”這篇文章就分享到這里了，希望以上內容可以對大家有一定的幫助，讓大家可以學到更多知識，如果覺得文章不錯，可以把它分享出去讓更多的人看到吧！

正文完