2. 程式人生 > >說說MQ之RocketMQ（二）

說說MQ之RocketMQ（二）

阿新 • • 發佈：2018-12-14

RocketMQ 的 Java API

RocketMQ 是用 Java 語言開發的，因此，其 Java API 相對是比較豐富的，當然也有部分原因是 RocketMQ 本身提供的功能就比較多。RocketMQ API 提供的功能包括，

  1. 廣播消費，這個在之前已經提到過；
  2. 訊息過濾，支援簡單的 Message Tag 過濾，也支援按 Message Header、body 過濾；
  3. 順序消費和亂序消費，之前也提到過，這裡的順序消費應該指的是普通順序性，這一點與 Kafka 相同；
  4. Pull 模式消費，這個是相對 Push 模式來說的，Kafka 就是 Pull 模式消費；
  5. 事務訊息，這個好像沒有開源，但是 example 程式碼中有示例，總之，不推薦用；
  6. Tag，RocketMQ 在 Topic 下面又分了一層 Tag，用於表示訊息類別，可以用來過濾，但是順序性還是以 Topic 來看；

單看功能的話，即使不算事務訊息，也不算 Tag，RocketMQ 也遠超 Kafka，Kafka 應該只實現了 Pull 模式消費 + 順序消費這2個功能。RocketMQ 的程式碼示例在 rocketmq-example 中，注意，程式碼是不能直接執行的，因為所有的程式碼都少了設定 name server 的部分，需要自己手動加上，例如，producer.setNamesrvAddr("192.168.232.23:9876");。

先來看一下生產者的 API，比較簡單，只有一種，如下，

import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.client.producer.DefaultMQProducer;
import com.alibaba.rocketmq.client.producer.MessageQueueSelector;
import com.alibaba.rocketmq.client.producer.SendResult;
import com.alibaba.rocketmq.common.message.Message;
import com.alibaba.rocketmq.common.message.MessageQueue;
import java.util.List;
public class Producer {
    public static void main(String[] args) throws MQClientException, InterruptedException {
        DefaultMQProducer producer = new DefaultMQProducer("ProducerGroupName");
        producer.setNamesrvAddr("192.168.232.23:9876");
        producer.start();
        for (int i = 0; i < 10; i++)
            try {
                {
                    Message msg = new Message("TopicTest1",// topic
                        "TagA",// tag
                        "OrderID188",// key
                        ("RocketMQ "+String.format("%05d", i)).getBytes());// body
                    SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                        @Override
                        public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                            Integer id = (Integer) arg;
                            int index = id % mqs.size();
                            return mqs.get(index);
                        }
                    }, i));
                    System.out.println(String.format("%05d", i)+sendResult);
                }
            }
            catch (Exception e) {
                e.printStackTrace();
            }
        producer.shutdown();
    }
}

可以發現，相比 Kafka 的 API，只多了 Tag，但實際上行為有很大不同。Kafka 的生產者客戶端，有同步和非同步兩種模式，但都是阻塞模式，send 方法返回傳送狀態的 Future，可以通過 Future 的 get 方法阻塞獲得傳送狀態。而 RocketMQ 採用的是同步非阻塞模式，傳送之後立刻返回傳送狀態（而不是 Future）。正常情況下，兩者使用上差別不大，但是在高可用場景中發生主備切換的時候，Kafka 的同步可以等待切換完成並重連，最後返回；而 RocketMQ 只能立刻報錯，由生產者選擇是否重發。所以，在生產者的 API 上，其實 Kafka 是要強一些的。

另外，RocketMQ 可以通過指定 MessageQueueSelector 類的實現來指定將訊息傳送到哪個分割槽去，Kafka 是通過指定生產者的 partitioner.class 引數來實現的，靈活性上 RocketMQ 略勝一籌。

再來看消費者的API，由於 RocketMQ 的功能比較多，我們先看 Pull 模式消費的API，如下，

import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import com.alibaba.rocketmq.client.consumer.DefaultMQPullConsumer;
import com.alibaba.rocketmq.client.consumer.PullResult;
import com.alibaba.rocketmq.client.consumer.store.OffsetStore;
import com.alibaba.rocketmq.client.consumer.store.RemoteBrokerOffsetStore;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.message.MessageExt;
import com.alibaba.rocketmq.common.message.MessageQueue;
public class PullConsumer {
    private static final Map<MessageQueue, Long> offseTable = new HashMap<MessageQueue, Long>();
    public static void main(String[] args) throws MQClientException {
        DefaultMQPullConsumer consumer = new DefaultMQPullConsumer("please_rename_unique_group_name_5");
        consumer.setNamesrvAddr("192.168.232.23:9876");
        consumer.start();
        Set<MessageQueue> mqs = consumer.fetchSubscribeMessageQueues("TopicTest1");
        for (MessageQueue mq : mqs) {
            System.out.println("Consume from the queue: " + mq);
            SINGLE_MQ: while (true) {
                try {
                    long offset = consumer.fetchConsumeOffset(mq, true);
                    PullResult pullResult =
                            consumer.pullBlockIfNotFound(mq, null, getMessageQueueOffset(mq), 32);
                    if (null != pullResult.getMsgFoundList()) {
                        for (MessageExt messageExt : pullResult.getMsgFoundList()) {
                            System.out.print(new String(messageExt.getBody()));
                            System.out.print(pullResult);
                            System.out.println(messageExt);
                        }
                    }
                    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)
            return offset;
        return 0;
    }
}

這部分的 API 其實是和 Kafka 很相似的，唯一不同的是，RocketMQ 需要手工管理 offset 和指定分割槽，而 Kafka 可以自動管理（當然也可以手動管理），並且不需要指定分割槽（分割槽是在 Kafka 訂閱的時候指定的）。例子中，RocketMQ 使用 HashMap 自行管理，也可以用 OffsetStore 介面，提供了兩種管理方式，本地檔案和遠端 Broker。這部分感覺兩者差不多。

下面再看看 Push 模式順序消費，程式碼如下，

import java.util.List;
import java.util.concurrent.atomic.AtomicLong;
import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerOrderly;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.MessageExt;
public class Consumer {
    public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_3");
        consumer.setNamesrvAddr("192.168.232.23:9876");
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
        consumer.subscribe("TopicTest1", "TagA || TagC || TagD");
        consumer.registerMessageListener(new MessageListenerOrderly() {
            AtomicLong consumeTimes = new AtomicLong(0);
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                context.setAutoCommit(false);
                System.out.println(Thread.currentThread().getName() + " Receive New Messages: " + msgs);
                this.consumeTimes.incrementAndGet();
                if ((this.consumeTimes.get() % 2) == 0) {
                    return ConsumeOrderlyStatus.SUCCESS;
                }
                else if ((this.consumeTimes.get() % 3) == 0) {
                    return ConsumeOrderlyStatus.ROLLBACK;
                }
                else if ((this.consumeTimes.get() % 4) == 0) {
                    return ConsumeOrderlyStatus.COMMIT;
                }
                else if ((this.consumeTimes.get() % 5) == 0) {
                    context.setSuspendCurrentQueueTimeMillis(3000);
                    return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
        consumer.start();
        System.out.println("Consumer Started.");
    }
}

雖然提供了 Push 模式，RocketMQ 內部實際上還是 Pull 模式的 MQ，Push 模式的實現應該採用的是長輪詢，這點與 Kafka 一樣。使用該方式有幾個注意的地方，

  1. 接收訊息的監聽類要使用 MessageListenerOrderly；
  2. ConsumeFromWhere 有幾個引數，表示從頭開始消費，從尾開始消費，還是從某個 TimeStamp 開始消費；
  3. 可以控制 offset 的提交，應該就是 context.setAutoCommit(false); 的作用；

控制 offset 提交這個特性非常有用，某種程度上擴充套件一下，就可以當做事務來用了，看程式碼 ConsumeMessageOrderlyService 的實現，其實並沒有那麼複雜，在不啟用 AutoCommit 的時候，只有返回 COMMIT 才 commit offset；啟用 AutoCommit 的時候，返回 COMMIT、ROLLBACK（這個比較扯）、SUCCESS 的時候，都 commit offset。

後來發現，commit offset 功能在 Kafka 裡面也有提供，使用新的 API，呼叫 consumer.commitSync。

再看一個 Push 模式亂序消費 + 訊息過濾的例子，消費者的程式碼如下，

import java.util.List;
import com.alibaba.rocketmq.client.consumer.DefaultMQPushConsumer;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import com.alibaba.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import com.alibaba.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import com.alibaba.rocketmq.client.exception.MQClientException;
import com.alibaba.rocketmq.common.message.MessageExt;
public class Consumer {
    public static void main(String[] args) throws InterruptedException, MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("ConsumerGroupNamecc4");
        consumer.setNamesrvAddr("192.168.232.23:9876");
        consumer.subscribe("TopicTest1", MessageFilterImpl.class.getCanonicalName());
        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.");
    }
}

這個例子與之前順序消費不同的地方在於，

  1. 接收訊息的監聽類使用的是 MessageListenerConcurrently；
  2. 回撥方法中，使用的是自動 offset commit；
  3. 訂閱的時候增加了訊息過濾類 MessageFilterImpl；

訊息過濾類 MessageFilterImpl 的程式碼如下，

import com.alibaba.rocketmq.common.filter.MessageFilter;
import com.alibaba.rocketmq.common.message.MessageExt;
public class MessageFilterImpl implements MessageFilter {
    @Override
    public boolean match(MessageExt msg) {
        String property = msg.getUserProperty("SequenceId");
        if (property != null) {
            int id = Integer.parseInt(property);
            if ((id % 3) == 0 && (id > 10)) {
                return true;
            }
        }
        return false;
    }
}

RocketMQ 執行過濾是在 Broker 端，Broker 所在的機器會啟動多個 FilterServer 過濾程序；Consumer 啟動後，會向 FilterServer 上傳一個過濾的 Java 類；Consumer 從 FilterServer 拉訊息，FilterServer 將請求轉發給 Broker，FilterServer 從 Broker 收到訊息後，按照 Consumer 上傳的 Java 過濾程式做過濾，過濾完成後返回給 Consumer。這種過濾方法可以節省網路流量，但是增加了 Broker 的負擔。可惜我沒有實驗出來使用過濾的效果，即使是用 github wiki 上的例子8也沒成功，不糾結了。RocketMQ 的按 Tag 過濾的功能也是在 Broker 上做的過濾，能用，是個很方便的功能。

還有一種廣播消費模式，比較簡單，可以去看程式碼，不再列出。

總之，RocketMQ 提供的功能比較多，比 Kafka 多很多易用的 API。

相關推薦

說說MQ之RocketMQ（二）

RocketMQ 的 Java API RocketMQ 是用 Java 語言開發的，因此，其 Java API 相對是比較豐富的，當然也有部分原因是 RocketMQ 本身提供的功能就比較多。RocketMQ API 提供的功能包括， 廣播消費，這個在之前已經提到過；

說說MQ之RocketMQ（一）

RocketMQ 是出自 A 公司的開源產品，用 Java 語言實現，在設計時參考了 Kafka，並做出了自己的一些改進，訊息可靠性上比 Kafka 更好，目前，RocketMQ 的文件仍然不夠豐富 1 2，社群仍然無法與 Kafka 比肩，但 A 公司已經推出了基於 Ro

說說MQ之RocketMQ（三）

RocketMQ 的主備模式 按之前所說，只有 RocketMQ 的多主多從非同步複製是可以生產使用的，因此只在這個場景下測試。另外，訊息採用 Push 順序模式消費。 假設叢集採用2主2備的模式，需要啟動4個 Broker，配置檔案如下， brokerName=broke

深入學習之mysql（二）表的操作

uniq order fault change incr 相關 約束 設置 type 1、表：是數據庫中的存儲數據的基本單位，一個表包含若幹個字段和值 2、創建表： 　　CREATE TABLE 表名稱 ( 　　字段名1　　數據庫類型1　　[約束條件1], 　　字段名2　　

自動化運維之saltstack（二）states介紹及使用

配置文件 如何 states master 根目錄 一、什麽是Salt States？Salt States是Salt模塊的擴展，主系統使用的狀態系統叫SLS系統，SLS代表Saltstack State，Salt是一些狀態文件，其中包含有關如何配置Salt子節點的信息，這些狀態被存放在一

Spring MVC 的 研發之路 （二）

eight 對象 files tin servlet映射 資源 研發 領域 延遲 二、web.xml的簡單配置介紹1 1、啟動Web項目時，容器回去讀web.xml配置文件裏的兩個節點<context-param>和<listener

Java面向對象編程之繼承（二）

抽象方法 oid 修飾 屬於 ... 方法的參數 要求 ring 覆蓋 在上一篇博客中，我們已經了解繼承的基本概念和語法，那麽今天我們就來聊一聊有關於繼承的其他東西。 讓我們來了解一下什麽是方法重載（overload）和方法覆蓋（override） 方法重載（overloa

SIPp常用腳本之UAS（二）

invite pcm option 響應 etime aud 服務器 esp contact 看名字就能猜出來，這是作為SIP消息服務端的存在，啟動uas，等著接受SIP消息並且給出響應。 一、uas.xml <?xml version="2.0" encoding

Java學習之路（二）流程控制語句

循環 cas 學習之路 將不 乘法表 length 跳出循環 spa int if、if…else…語句 if (true) { System.out.println("為真時執行");

二次元碼農的成長之路（二）json到底有什麽用途

頁面 asc 語法 處理 交互 分隔 規則 stl 成長之路 寫的有問題請指出 一、什麽是json 1json指的是javaScript的表示方法 2JSON是輕量級的文本數據交換格式 3 Json是獨立語言 4 json具有自我描述性 更易理解 二、語法規則 1對

Hibernate框架之路（二）一對多關系

true pre conn 關系 編寫 none isp exce dia 在下列案例中： 　　客戶是一；聯系人是多；客戶可以擁有多個聯系人；一個聯系人只有一個客戶 　　1、編寫實體類 package com.pojo; import java.util

Hibernate框架之路（二）多對多多關系

-m let ble ssi gin rop assert on() odi 1、配置實體類 package cn.pojo; import java.util.HashSet; import java.util.Set; public class

基礎篇之集合（二）總結

線程不安全 emp abstract 和集 write next 不可變 叠代器 關系 1. Map集合和collection結合的區別 1》Collection一次存一個元素；Map一次存一對元素； 2》Collection是單列集合；Map是雙列集合； 3》Map中的存

JavaScript的進階之路（二）函數簡介，變量、作用域和內存問題

ret 優化 person get 簡介 web瀏覽器 都是 add 是把 <h3>ECMAScript中函數不存在函數簽名的概念，沒有重載</h3><h3>無需指定返回值，可以在任何時候返回任何值。未指定返回值的函數，返回的是一個特殊

小白的linux學習之旅（二）

探索linux一、linux系統結構linux是一個倒樹結構linux中所有的東西都是文件這些文件都在系統頂級目錄“/” /就是根目錄/目錄以下為二級目錄這些目錄都是系統裝機時系統自動建立的二級目錄的作用/bin 二進制可執行文件也就是系統命令/sbin

java集合類源碼分析之List（二）

頻繁 null 並且 reel closed tco 默認 java集合 進行 這一節主要介紹List接口的幾個實現類的區別： 1.線程安全 Vector是線程安全的，而ArrayList和LinkedList是非線程安全的。從源碼中我們可知，Vector類中的方法大部分

(轉）JavaWeb學習之Servlet（二）----Servlet的生命周期、繼承結構、修改Servlet模板

ext 核心 system 網頁 基本 麻煩 extends 用戶名 servlet對象 【聲明】 歡迎轉載，但請保留文章原始出處→_→ 文章來源：http://www.cnblogs.com/smyhvae/p/4140466.html 一、http協議回顧： 在上一

Ajax技術之XMLHttpRequest（二）【XMLHttpRequest常用方法和屬性】

tro 沒有 body htm 事件 thead 指定 response date 一、XMLHttpRequest中常用的方法： （1）open（）方法：用於設置進行異步請求目標的URL、請求方法以及其他參數信息。 函數原型：open("method","URL",asy

我的C#跨平臺之旅（二）：開發一組標準的Restful API

ref 運行 mar margin bruce ora soft left 啟用 添加NuGet引用：Microsoft.AspNet.WebApi.Owin 在啟動類啟用WebApi； 添加一個Controller類，代碼如下： 運行程序

python學習之路（二） -- 函數、JSON、終端樣式

blog ade def 數量 通過 等於 name tuple args 函數 函數構成 定義函數:使用def即可 def __getName(idCard): return user_info[idCard].Name 其中，__get