輕鬆搞定RocketMQ入門

来源:https://www.cnblogs.com/AIPAOJIAO/archive/2018/08/09/9452242.html
-Advertisement-
Play Games

RocketMQ是一款分散式、隊列模型的消息中間件，具有以下特點： 能夠保證嚴格的消息順序 提供豐富的消息拉取模式 高效的訂閱者水平擴展能力 實時的消息訂閱機制 億級消息堆積能力 RocketMQ網路部署特 （1）NameServer是一個幾乎無狀態的節點，可集群部署，節點之間無任何信息同步 （2） ...


RocketMQ是一款分散式、隊列模型的消息中間件，具有以下特點：

  1. 能夠保證嚴格的消息順序

  2. 提供豐富的消息拉取模式

  3. 高效的訂閱者水平擴展能力

  4. 實時的消息訂閱機制

  5. 億級消息堆積能力

RocketMQ網路部署特

（1）NameServer是一個幾乎無狀態的節點，可集群部署，節點之間無任何信息同步

（2）Broker部署相對複雜，Broker氛圍Master與Slave，一個Master可以對應多個Slaver，但是一個Slaver只能對應一個Master，Master與Slaver的對應關係通過指定相同的BrokerName，不同的BrokerId來定義，BrokerId為0表示Master，非0表示Slaver。Master可以部署多個。每個Broker與NameServer集群中的所有節點建立長連接，定時註冊Topic信息到所有的NameServer

（3）Producer與NameServer集群中的其中一個節點（隨機選擇）建立長連接，定期從NameServer取Topic路由信息，並向提供Topic服務的Master建立長連接，且定時向Master發送心跳。Produce完全無狀態，可集群部署

（4）Consumer與NameServer集群中的其中一個節點（隨機選擇）建立長連接，定期從NameServer取Topic路由信息，並向提供Topic服務的Master、Slaver建立長連接，且定時向Master、Slaver發送心跳。Consumer即可從Master訂閱消息，也可以從Slave訂閱消息，訂閱規則由Broker配置決定

 

RocketMQ儲存特點

（1）零拷貝原理：Consumer消費消息過程，使用了零拷貝，零拷貝包括一下2中方式，RocketMQ使用第一種方式，因小塊數據傳輸的要求效果比sendfile方式好

a )使用mmap+write方式

優點：即使頻繁調用，使用小文件塊傳輸，效率也很高

缺點：不能很好的利用DMA方式，會比sendfile多消耗CPU資源，記憶體安全性控制複雜，需要避免JVM Crash問題

b）使用sendfile方式

優點：可以利用DMA方式，消耗CPU資源少，大塊文件傳輸效率高，無記憶體安全新問題

缺點：小塊文件效率低於mmap方式，只能是BIO方式傳輸，不能使用NIO

（2）數據存儲結構

RocketMQ關鍵特性

1.單機支持1W以上的持久化隊

（1）所有數據單獨儲存到commit Log ，完全順序寫，隨機讀

（2）對最終用戶展現的隊列實際只儲存消息在Commit Log 的位置信息，並且串列方式刷盤

這樣做的好處：

（1）隊列輕量化，單個隊列數據量非常少

（2）對磁碟的訪問串列話，避免磁碟競爭，不會因為隊列增加導致IOWait增高

每個方案都有優缺點，他的缺點是：

（1）寫雖然是順序寫，但是讀卻變成了隨機讀

（2）讀一條消息，會先讀Consume Queue，再讀Commit Log，增加了開銷

（3）要保證Commit Log 與 Consume Queue完全的一致，增加了編程的複雜度

以上缺點如何客服：

（1）隨機讀，儘可能讓讀命中pagecache，減少IO操作，所以記憶體越大越好。如果系統中堆積的消息過多，讀數據要訪問硬碟會不會由於隨機讀導致系統性能急劇下降，答案是否定的。

a）訪問pagecache時，即使只訪問1K的消息，系統也會提前預讀出更多的數據，在下次讀時就可能命中pagecache

b）隨機訪問Commit Log 磁碟數據，系統IO調度演算法設置為NOOP方式，會在一定程度上將完全的隨機讀變成順序跳躍方式，而順序跳躍方式讀較完全的隨機讀性能高5倍

（2）由於Consume Queue存儲數量極少，而且順序讀，在pagecache的與讀取情況下，Consume Queue的讀性能與記憶體幾乎一直，即使堆積情況下。所以可以認為Consume Queue完全不會阻礙讀性能

（3）Commit Log中存儲了所有的元信息，包含消息體，類似於MySQl、Oracle的redolog，所以只要有Commit Log存在， Consume Queue即使丟失數據，仍可以恢復出來

2.刷盤策略

rocketmq中的所有消息都是持久化的，先寫入系統pagecache，然後刷盤，可以保證記憶體與磁碟都有一份數據，訪問時，可以直接從記憶體讀取

2.1非同步刷盤

在有 RAID 卡， SAS 15000 轉磁碟測試順序寫文件，速度可以達到 300M 每秒左右，而線上的網卡一般都為千兆網卡，寫磁碟速度明顯快於數據網路入口速度，那麼是否可以做到寫完 記憶體就向用戶返回，由後臺線程刷盤呢？

(1). 由於磁碟速度大於網卡速度，那麼刷盤的進度肯定可以跟上消息的寫入速度。

(2). 萬一由於此時系統壓力過大，可能堆積消息，除了寫入 IO，還有讀取 IO，萬一齣現磁碟讀取落後情況，會不會導致系統記憶體溢出，答案是否定的，原因如下：

a) 寫入消息到 PAGECACHE 時，如果記憶體不足，則嘗試丟棄乾凈的 PAGE，騰出記憶體供新消息使用，策略是 LRU 方式。

b) 如果乾凈頁不足，此時寫入 PAGECACHE 會被阻塞，系統嘗試刷盤部分數據，大約每次嘗試 32 個 PAGE，來找出更多乾凈 PAGE。
綜上，記憶體溢出的情況不會出現

2.2同步刷盤：

同步刷盤與非同步刷盤的唯一區別是非同步刷盤寫完 PAGECACHE 直接返回，而同步刷盤需要等待刷盤完成才返回，同步刷盤流程如下：

（1）寫入 PAGECACHE 後，線程等待，通知刷盤線程刷盤。

（2）刷盤線程刷盤後，喚醒前端等待線程，可能是一批線程。

（3）前端等待線程向用戶返回成功。

3.消息查詢

3.1按照MessageId查詢消息

MsgId總共16個位元組，包含消息儲存主機地址，消息Commit Log Offset。從MsgId中解析出Broker的地址和Commit Log 偏移地址，然後按照存儲格式所在位置消息buffer解析成一個完整消息

3.2按照Message Key查詢消息

1.根據查詢的key的hashcode%slotNum得到具體的槽位置 （slotNum是一個索引文件裡面包含的最大槽目數目，例如圖中所示slotNum=500W）

2.根據slotValue（slot對應位置的值）查找到索引項列表的最後一項（倒序排列，slotValue總是指向最新的一個索引項）

3.遍歷索引項列表返回查詢時間範圍內的結果集（預設一次最大返回的32條記錄）

4.Hash衝突，尋找key的slot位置時相當於執行了兩次散列函數，一次key的hash，一次key的hash取值模，因此這裡存在兩次衝突的情況；第一種，key的hash值不同但模數相同，此時查詢的時候會在比較第一次key的hash值（每個索引項保存了key的hash值），過濾掉hash值不想等的情況。第二種，hash值相等key不想等，出於性能的考慮衝突的檢測放到客戶端處理（key的原始值是存儲在消息文件中的，避免對數據文件的解析），客戶端比較一次消息體的key是否相同

5.存儲，為了節省空間索引項中存儲的時間是時間差值（存儲時間——開始時間，開始時間存儲在索引文件頭中），整個索引文件是定長的，結構也是固定的

4.伺服器消息過濾

RocketMQ的消息過濾方式有別於其他的消息中間件，是在訂閱時，再做過濾，先來看下Consume Queue存儲結構

1.在Broker端進行Message Tag比較，先遍歷Consume Queue，如果存儲的Message Tag與訂閱的Message Tag不符合，則跳過，繼續比對下一個，符合則傳輸給Consumer。註意Message Tag是字元串形式，Consume Queue中存儲的是其對應的hashcode，比對時也是比對hashcode

2.Consumer收到過濾消息後，同樣也要執行在broker端的操作，但是比對的是真實的Message Tag字元串，而不是hashcode

為什麼過濾要這麼做？

1.Message Tag存儲hashcode，是為了在Consume Queue定長方式存儲，節約空間

2.過濾過程中不會訪問Commit Log 數據，可以保證堆積情況下也能高效過濾

3.即使存在hash衝突，也可以在Consumer端進行修正，保證萬無一失

5.單個JVM進程也能利用機器超大記憶體

1.Producer發送消息，消息從socket進入java 堆

2.Producer發送消息，消息從java堆進入pagecache，物理記憶體

3.Producer發送消息，由非同步線程刷盤，消息從pagecache刷入磁碟

4.Consumer拉消息（正常消費），消息直接從pagecache（數據在物理記憶體）轉入socket，到達Consumer，不經過java堆。這種消費場景最多，線上96G物理記憶體，按照1K消息算，可以物理緩存1億條消息

5.Consumer拉消息（異常消費），消息直接從pagecache轉入socket

6.Consumer拉消息（異常消費），由於socket訪問了虛擬記憶體，產生缺頁中斷，此時會產生磁碟IO，從磁碟Load消息到pagecache，然後直接從socket發出去

7.同5

8.同6

6.消息堆積問題解決辦法

1 消息的堆積容量、依賴磁碟大小

2 發消息的吞吐量大小受影響程度、無Slave情況，會受一定影響、有Slave情況，不受影響

3 正常消費的Consumer是否會受影響、無Slave情況，會受一定影響、有Slave情況，不受影響

4 訪問堆積在磁碟的消息時，吞吐量有多大、與訪問的併發有關，最終會降到5000左右

在有Slave情況下，Master一旦發現Consumer訪問堆積在磁碟的數據時，回想Consumer下達一個重定向指令，令Consumer從Slave拉取數據，這樣正常的發消息與正常的消費不會因為堆積受影響，因為系統將堆積場景與非堆積場景分割在了兩個不同的節點處理。這裡會產生一個問題，Slave會不會寫性能下降，答案是否定的。因為Slave的消息寫入只追求吞吐量，不追求實時性，只要整體的吞吐量高就行了，而Slave每次都是從Master拉取一批數據，如1M，這種批量順序寫入方式使堆積情況，整體吞吐量影響相對較小，只是寫入RT會變長。

服務端安裝部署

我是在虛擬機中的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位置。

 

 

文章有點長，大家覺得作者總結的還可以，大家可以點擊下方二維碼進行關註。《Java爛豬皮》公眾號聊的不僅僅是Java技術知識，還有面試等乾貨，後期還有大量架構乾貨。大家一起關註吧！關註爛豬皮，你會瞭解的更多..............

 


您的分享是我們最大的動力!

-Advertisement-
Play Games
更多相關文章
  • 第二十四節：Java語言基礎-講解數組的綜合應用
    數組的綜合應用 冒泡排序 為什麼這個排序要叫冒泡呢？為什麼不叫其他名詞呢？ 其實這個取名是根據排序演算法的基本思路命名的，見名知意，冒泡排序，就是想泡泡在水裡一樣，在水裡大的泡泡先浮出水面，就是大的先排出來，最小的最慢排出。 冒泡排序，是對排序的各個元素從頭到尾依次進行相鄰的大小比較， 比如你是隊長， ...
  • 【leetcode 簡單】第十四題 最後一個單詞的長度
    給定一個僅包含大小寫字母和空格 ' ' 的字元串，返回其最後一個單詞的長度。 如果不存在最後一個單詞，請返回 0 。 說明：一個單詞是指由字母組成，但不包含任何空格的字元串。 示例: 輸入: "Hello World" 輸出: 5 給定一個僅包含大小寫字母和空格 ' ' 的字元串，返回其最後一個單詞 ...
  • 【leetcode 簡單】第十三題 最大子序和
    給定一個整數數組 nums ，找到一個具有最大和的連續子數組（子數組最少包含一個元素），返回其最大和。 示例: 輸入: [-2,1,-3,4,-1,2,1,-5,4], 輸出: 6 解釋: 連續子數組 [4,-1,2,1] 的和最大，為 6。 class Solution: def maxSubAr ...
  • Mybatis的SqlSession運行原理
    前言 SqlSession是Mybatis最重要的構建之一，可以簡單的認為Mybatis一系列的配置目的是生成類似 JDBC生成的Connection對象的SqlSession對象，這樣才能與資料庫開啟“溝通”，通過SqlSession可以實現增刪改查（當然現在更加推薦是使用Mapper介面形式）， ...
  • 經典傅里葉演算法小集合 附完整c代碼
    前面寫過關於傅里葉演算法的應用例子。 《基於傅里葉變換的音頻重採樣演算法 (附完整c代碼)》 當然也就是舉個例子，主要是學習傅里葉變換。 這個重採樣思路還有點瑕疵， 稍微改一下，就可以支持多通道，以及提升性能。 當然思路很簡單，就是切分，合併。 留個作業哈。 本文不講過多的演算法思路，傅里葉變換的各種變種 ...
  • Consul+Nginx+Upsync+Linux+Keepalived+Lvs的動態負載均衡
    註： Nginx版本必須大於等於1.9，linux我使用的是7.0版本，記得關閉防火牆 開始正文： 這裡只提一下兩個比較會出錯的配置，consul，upsync，PCRE庫，SSL庫，ZLIB庫的配置網上很多不一一詳解（我是在/usr/local中配置的） Nginx的配置: cd /usr/loc ...
  • java關鍵字 volatile的作用及使用說明
    先來看看這個關鍵字是什麼意思：volatile [ˈvɒlətaɪl] adj. 易變的，不穩定的; 從翻譯上來看，volatile表示這個關鍵字是極易發生改變的。volatile是java語言中，最輕量級的併發同步機制。這個關鍵字有如下兩個作用：1、任何對volatile變數的修改，java中的其 ...
  • Django---Models
    資料庫的配置 1 django預設支持sqlite，mysql, oracle,postgresql資料庫。 <1> sqlite django預設使用sqlite的資料庫，預設自帶sqlite的資料庫驅動 , 引擎名稱：django.db.backends.sqlite3 <2> mysql 引擎 ...
一周排行
    -Advertisement-
    Play Games
  • 移動開發（一）：使用.NET MAUI開發第一個安卓APP
    移動開發（一）：使用.NET MAUI開發第一個安卓APP 對於工作多年的C#程式員來說，近來想嘗試開發一款安卓APP，考慮了很久最終選擇使用.NET MAUI這個微軟官方的框架來嘗試體驗開發安卓APP，畢竟是使用Visual Studio開發工具，使用起來也比較的順手，結合微軟官方的教程進行了安卓 ...
  • wpf ToggleButton選中效果和一個登錄界面
    前言 QuestPDF 是一個開源 .NET 庫，用於生成 PDF 文檔。使用了C# Fluent API方式可簡化開發、減少錯誤並提高工作效率。利用它可以輕鬆生成 PDF 報告、發票、導出文件等。 項目介紹 QuestPDF 是一個革命性的開源 .NET 庫，它徹底改變了我們生成 PDF 文檔的方 ...
  • 彈幕樹洞項目功能新增篇
    項目地址 項目後端地址： https://github.com/ZyPLJ/ZYTteeHole 項目前端頁面地址： ZyPLJ/TreeHoleVue (github.com) https://github.com/ZyPLJ/TreeHoleVue 目前項目測試訪問地址： http://tree ...
  • 第27篇 sqlserver2022詳細安裝步驟
    話不多說，直接開乾 一.下載 1.官方鏈接下載： https://www.microsoft.com/zh-cn/sql-server/sql-server-downloads 2.在下載目錄中找到下麵這個小的安裝包 SQL2022-SSEI-Dev.exe，運行開始下載SQL server； 二. ...
  • .NET 開源高性能 MQTT 類庫
    前言 隨著物聯網（IoT）技術的迅猛發展，MQTT（消息隊列遙測傳輸）協議憑藉其輕量級和高效性，已成為眾多物聯網應用的首選通信標準。 MQTTnet 作為一個高性能的 .NET 開源庫，為 .NET 平臺上的 MQTT 客戶端與伺服器開發提供了強大的支持。 本文將全面介紹 MQTTnet 的核心功能 ...
  • Serilog文檔翻譯系列（六） - 可用的接收器、增強器、格式化輸出
    Serilog支持多種接收器用於日誌存儲，增強器用於添加屬性，LogContext管理動態屬性，支持多種輸出格式包括純文本、JSON及ExpressionTemplate。還提供了自定義格式化選項，適用於不同需求。 ...
  • 警惕 Visual Studio 屬性求值副作用導致邏輯不符合預期
    目錄簡介獲取 HTML 文檔解析 HTML 文檔測試參考文章 簡介 動態內容網站使用 JavaScript 腳本動態檢索和渲染數據，爬取信息時需要模擬瀏覽器行為，否則獲取到的源碼基本是空的。 本文使用的爬取步驟如下： 使用 Selenium 獲取渲染後的 HTML 文檔 使用 HtmlAgility ...
  • [使用目前最新版]HybridCLR6.9.0+YooAsset2.2.4實現純C# Unity熱更新方案 (一)
    1.前言 什麼是熱更新 游戲或者軟體更新時，無需重新下載客戶端進行安裝，而是在應用程式啟動的情況下，在內部進行資源或者代碼更新 Unity目前常用熱更新解決方案 HybridCLR，Xlua，ILRuntime等 Unity目前常用資源管理解決方案 AssetBundles，Addressable， ...
  • 在 ASP.NET Core Web API 中使用操作篩選器統一處理通用操作
    本文章主要是在C# ASP.NET Core Web API框架實現向手機發送驗證碼簡訊功能。這裡我選擇是一個互億無線簡訊驗證碼平臺，其實像阿裡雲，騰訊雲上面也可以。 首先我們先去 互億無線 https://www.ihuyi.com/api/sms.html 去註冊一個賬號 註冊完成賬號後，它會送 ...
  • 第28篇 如何.net中實現高效可靠數據同步api
    通過以下方式可以高效，並保證數據同步的可靠性 1.API設計 使用RESTful設計，確保API端點明確，並使用適當的HTTP方法（如POST用於創建，PUT用於更新）。 設計清晰的請求和響應模型，以確保客戶端能夠理解預期格式。 2.數據驗證 在伺服器端進行嚴格的數據驗證，確保接收到的數據符合預期格 ...
所有分類