2. 程式人生 > >RocketMQ——順序消費（程式碼）

RocketMQ——順序消費（程式碼）

阿新 • • 發佈：2019-01-08

本部落格主要是以程式碼示例來了解順序消費的相關內容,建議在此之前先了解下順序消費的原理。

注：RocketMQ可以嚴格的保證訊息有序，但這個順序，不是全域性順序，只是分割槽（queue）順序，如果想要全域性順序，那麼需要保證只有一個分割槽。

順序消費簡介

1.普通順序消費

順序消費的一種，正常情況下可以保證完全的順序訊息，但是一旦發生通訊異常，Broker重啟，由於佇列總數法還是能變化，雜湊取模後定位的佇列會變化，產生短暫的訊息順序不一致。

2.嚴格順序訊息

順序訊息的一種，無論正常異常情況都能保證順序，但是犧牲了分散式failover特性，即broker叢集中要有一臺機器不可用，則整個叢集都不可用，服務可用性大大降低。如果伺服器部署為同步雙寫模式，此缺陷可通過備機自動切換為主避免，不過仍然會存在幾分鐘的服務不可用。

目前已知的應用只有資料庫binlog同步強依賴嚴格順序訊息，其他應用絕大部分都可以容忍短暫亂序，推薦使用普通的順序訊息。

1.producer

package com.gwd.rocketmq;

import java.io.IOException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;

import com.alibaba.rocketmq.client.exception.MQBrokerException;
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 com.alibaba.rocketmq.remoting.exception.RemotingException;

/** 
* @FileName Producer.java
* @Description:
* @author gu.weidong
* @version V1.0
* @createtime 2018年7月3日 上午9:59:38 
* 修改歷史：
* 時間           作者          版本        描述
*====================================================  
*
*/
/**
 * Producer，傳送順序訊息
 */
public class Producer {
	
    public static void main(String[] args) throws IOException {
        try {
            DefaultMQProducer producer = new DefaultMQProducer("sequence_producer");
 
            producer.setNamesrvAddr("192.168.159.128:9876;192.168.159.129:9876");
 
            producer.start();
 
            String[] tags = new String[] { "TagA", "TagC", "TagD" };
            
            // 訂單列表
            List<OrderDO> orderList =  new Producer().buildOrders();
            
            Date date = new Date();
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
            String dateStr = sdf.format(date);
            for (int i = 0; i < 10; i++) {
                // 加個時間字尾
                String body = dateStr + " Hello RocketMQ " + orderList.get(i).getOrderId()+orderList.get(i).getDesc();
                Message msg = new Message("SequenceTopicTest", tags[i % tags.length], "KEY" + i, body.getBytes());
 
                SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                    @Override
                    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                        Long id = Long.valueOf((String)arg);
                        long index = id % mqs.size();
                        return mqs.get((int)index);
                    }
                }, orderList.get(i).getOrderId());//通過訂單id來獲取對應的messagequeue
 
                System.out.println(sendResult + ", body:" + body);
            }
            
            producer.shutdown();
 
        } catch (MQClientException e) {
            e.printStackTrace();
        } catch (RemotingException e) {
            e.printStackTrace();
        } catch (MQBrokerException e) {
            e.printStackTrace();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.in.read();
    }
    
    /**
     * 生成模擬訂單資料 
     */
    private List<OrderDO> buildOrders() {
    	List<OrderDO> orderList = new ArrayList<OrderDO>();
 
    	OrderDO OrderDO = new OrderDO();
        OrderDO.setOrderId("15103111039");
    	OrderDO.setDesc("建立");
    	orderList.add(OrderDO);
    	
    	OrderDO = new OrderDO();
    	OrderDO.setOrderId("15103111065");
    	OrderDO.setDesc("建立");
    	orderList.add(OrderDO);
    	
    	OrderDO = new OrderDO();
    	OrderDO.setOrderId("15103111039");
    	OrderDO.setDesc("付款");
    	orderList.add(OrderDO);
    	
    	OrderDO = new OrderDO();
    	OrderDO.setOrderId("15103117235");
    	OrderDO.setDesc("建立");
    	orderList.add(OrderDO);
    	
    	OrderDO = new OrderDO();
    	OrderDO.setOrderId("15103111065");
    	OrderDO.setDesc("付款");
    	orderList.add(OrderDO);
    	
    	OrderDO = new OrderDO();
    	OrderDO.setOrderId("15103117235");
    	OrderDO.setDesc("付款");
    	orderList.add(OrderDO);
    	
    	OrderDO = new OrderDO();
    	OrderDO.setOrderId("15103111065");
    	OrderDO.setDesc("完成");
    	orderList.add(OrderDO);
    	
    	OrderDO = new OrderDO();
    	OrderDO.setOrderId("15103111039");
    	OrderDO.setDesc("推送");
    	orderList.add(OrderDO);
    	
    	OrderDO = new OrderDO();
    	OrderDO.setOrderId("15103117235");
    	OrderDO.setDesc("完成");
    	orderList.add(OrderDO);
    	
    	OrderDO = new OrderDO();
    	OrderDO.setOrderId("15103111039");
    	OrderDO.setDesc("完成");
    	orderList.add(OrderDO);
    	return orderList;
    }
}
此處需要注意，producer.send(msg, new MessageQueueSelector()),如果需要全域性有序，只需要使new MessageQueueSelector().select(List<MessageQueue> mqs, Message msg, Object arg)方法返回值唯一且不變，例如：
SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                    @Override
                    public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {
                        Long id = Long.valueOf((String)arg);
                        long index = id % mqs.size();
                        return mqs.get((int)index);
                    }
                }, orderList.get(0).getOrderId());//通過訂單id來獲取對應的messagequeue

這邊獲取到的queue永遠都是唯一的且確定的（此處只是舉個簡單的例子，orderList.get(i).getOrderId()改為0亦可）

2.錯誤的Consumer

package com.gwd.rocketmq;

import java.util.List;
import java.util.Random;
import java.util.concurrent.TimeUnit;

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.consumer.ConsumeFromWhere;
import com.alibaba.rocketmq.common.message.MessageExt;

/** 
* @FileName WrongConsumer.java
* @Description:
* @author gu.weidong
* @version V1.0
* @createtime 2018年7月3日 下午3:13:16 
* 修改歷史：
* 時間           作者          版本        描述
*====================================================  
*
*/
public class WrongConsumer {
	public static void main(String[] args) throws MQClientException {
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("please_rename_unique_group_name_3");
        consumer.setNamesrvAddr("192.168.159.128:9876;192.168.159.129:9876");
        /**
         * 設定Consumer第一次啟動是從佇列頭部開始消費還是佇列尾部開始消費<br>
         * 如果非第一次啟動，那麼按照上次消費的位置繼續消費
         */
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
 
        consumer.subscribe("SequenceTopicTest", "TagA || TagC || TagD");
 
        consumer.registerMessageListener(new MessageListenerConcurrently() {
 
            Random random = new Random();
 
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                System.out.print(Thread.currentThread().getName() + " Receive New Messages: " );
                for (MessageExt msg: msgs) {
                    System.out.println(msg + ", content:" + new String(msg.getBody()));
                }
                try {
                    //模擬業務邏輯處理中...
                    TimeUnit.SECONDS.sleep(random.nextInt(10));
                } catch (Exception e) {
                    e.printStackTrace();
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
 
        consumer.start();
 
        System.out.println("Consumer Started.");
    }
}

注意：要想要有順序，那麼這邊吃監聽器就不能是MessageListenerConcurrently了，其顯示效果如下：


3.正確的Consumer

package com.gwd.rocketmq;

import java.util.List;
import java.util.Random;
import java.util.concurrent.TimeUnit;

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;

/** 
* @FileName Consumer.java
* @Description:
* @author gu.weidong
* @version V1.0
* @createtime 2018年7月3日 上午10:05:26 
* 修改歷史：
* 時間           作者          版本        描述
*====================================================  
*
*/
/**
 * 順序訊息消費，帶事務方式（應用可控制Offset什麼時候提交）
 */
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.159.128:9876;192.168.159.129:9876");
        /**
         * 設定Consumer第一次啟動是從佇列頭部開始消費還是佇列尾部開始消費<br>
         * 如果非第一次啟動，那麼按照上次消費的位置繼續消費
         */
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
 
        consumer.subscribe("SequenceTopicTest", "TagA || TagC || TagD");
 
        consumer.registerMessageListener(new MessageListenerOrderly() {
 
            Random random = new Random();
 
            @Override
            public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
                context.setAutoCommit(true);
                System.out.print(Thread.currentThread().getName() + " Receive New Messages: " );
                for (MessageExt msg: msgs) {
                    System.out.println(msg + ", content:" + new String(msg.getBody()));
                }
                try {
                    //模擬業務邏輯處理中...
                    TimeUnit.SECONDS.sleep(random.nextInt(10));
                } catch (Exception e) {
                    e.printStackTrace();
                }
                return ConsumeOrderlyStatus.SUCCESS;
            }
        });
 
        consumer.start();
        System.out.println("Consumer Started.");
    }
}

這邊的Consumer和上面的最明顯的區別在於對應的監聽器是MessageListenerOrderly，MessageListenerOrderly是能夠保證順序消費的。

顯示結果：


4.多個消費者

那如果有多個消費者呢？因為訊息傳送時被分配到多個佇列，這些佇列又會被分別傳送給消費者唯一消費，現在啟動兩個消費者，其消費情況如下圖：



結論：多個消費者時，各個消費者的訊息依舊是順序消費，且不會重複消費

原文轉自：https://blog.csdn.net/earthhour/article/details/78323026 ，在此基礎上部分程式碼略作修改

相關推薦

RocketMQ——順序消費（程式碼）

本部落格主要是以程式碼示例來了解順序消費的相關內容,建議在此之前先了解下順序消費的原理。注：RocketMQ可以嚴格的保證訊息有序，但這個順序，不是全域性順序，只是分割槽（queue）順序，如果想要全域性順序，那麼需要保證只有一個分割槽。順序消費簡介1.普通順序消費順序消費的

RocketMQ——Consumer篇：PULL模式下的訊息消費（DefaultMQPullConsumer）

1 應用層的使用方式 在應用層初始化DefaultMQPullConsumer類，然後呼叫該類的start方法啟動Consumer；接下來的消費步驟如下： 1、呼叫DefaultMQPullConsumer.fetchSubscribeMessageQueu

從遞迴（程式碼）的角度理解二叉樹的先序/中序/後序三種遍歷順序

二叉樹的遍歷原理其實是很簡單的，就是從二叉樹的根節點開始，把它的左節點以及左節點下面的節點再當作一棵二叉樹，和右節點以及右節點下面的節點也再當作一棵二叉樹，依此類推，直到節點下面沒有節點為止。二叉樹的有三種遍歷順序：先序遍歷、中序遍歷、後序遍歷，而這個順序都是以根節點作為參照

RocketMQ原始碼解析：Message拉取&消費（下）

title: RocketMQ 原始碼分析 —— Message 拉取與消費（下） date: 2017-05-11 tags: categories: RocketMQ permalink: RocketMQ/message-pull-and-cons

MySQL系列：基於binlog的增量訂閱與消費（一）

clas 需要 val tro ali cat tor rip 變化 　　在一些業務場景中，像在數據分析中我們有時候需要捕獲數據變化（CDC）；在數據審計中，我們也往往需要知道數據從這個點到另一個點的變化；同樣在實時分析中，我們有時候需要看到某個值得實時變化等。 要解決以上

Spring Cloud Spring Boot mybatis 企業分布式微服務雲（五）服務消費（Feign）【Dalston版】

host 微服務 https 完成 分布 tar blog 擴展 work Spring Cloud Feign Spring Cloud Feign是一套基於Netflix Feign實現的聲明式服務調用客戶端。它使得編寫Web服務客戶端變得更加簡單。我們只需要通過創建接

構建微服務架構Spring Cloud：服務消費（基礎）

成了 cloud framework shadow 即將 nbu 註冊中心 obj client 使用LoadBalancerClient 在Spring Cloud Commons中提供了大量的與服務治理相關的抽象接口，包括DiscoveryClient、這裏我們即將介紹

構建微服務架構Spring Cloud：服務消費（Ribbon）

架構 pid 編寫 動手 tap consumer pre 攔截器 over Spring Cloud Ribbon Spring Cloud Ribbon是基於Netflix Ribbon實現的一套客戶端負載均衡的工具。它是一個基於HTTP和TCP的客戶端負載均衡器。它可

構建微服務架構Spring Cloud：服務消費（Feign）

進行 string oca 成對 rest server 之前 int netflix Spring Cloud Feign Spring Cloud Feign是一套基於Netflix Feign實現的聲明式服務調用客戶端。它使得編寫Web服務客戶端變得更加簡單。我們只需

Spring Cloud 入門Eureka -Consumer服務消費（一）

ppi package AR con .so 1.8 Coding ng- int 這裏介紹：LoadBalancerClient接口，它是一個負載均衡客戶端的抽象定義，下面我們就看看如何使用Spring Cloud提供的負載均衡器客戶端接口來實現服務的消費。 引用之前的文

Spring Cloud構建微服務架構—服務消費（Ribbon）

ble DG 沒有 客戶 BE pla cati str 主類 Spring Cloud RibbonSpring Cloud Ribbon是基於Netflix Ribbon實現的一套客戶端負載均衡的工具。它是一個基於HTTP和TCP的客戶端負載均衡器。它可以通過在客戶端中

Spring Cloud構建微服務架構：服務消費（基礎）

消費 ring str frame emp default class a template pom.xml 使用LoadBalancerClient在Spring Cloud Commons中提供了大量的與服務治理相關的抽象接口，包括DiscoveryClient、這裏我

Spring Cloud構建PC蛋蛋源碼下載微服務架構：服務消費（基礎）

true ota ctu temp control lan prope 源碼下載 builder PC蛋蛋源碼下載論壇：haozbbs.com Q1446595067 使用LoadBalancerClient 在Spring Cloud Commons中提供了大量的與服務治

Oracle儲存過程例項分析總結（程式碼）

1.儲存過程結構 1.1 第一個儲存過程 ? 1 2 3 4

資料結構——表示式求值（程式碼）

表示式求值 C++ 環境codeblocks17 通過 /* 表示式求值，可用運算子 +-/*(){}[] @CGQ 2018/10/30 */ #include<stdio.h> #include<stdlib.h> #include<ctype.h>

資料結構——二叉樹（程式碼）

 二叉樹 C++ 環境codeblocks17 通過 /* 二叉樹 使用了自定義的 棧 和 佇列 @CGQ 2018/10/29 */ #include <iostream> #include <stdio.h> #include <stdlib.h&

資料結構——線索二叉樹（程式碼）

線索二叉樹 C++ 環境codeblocks17 通過 /* 線索二叉樹 @CGQ 2018/10/29 */ #include <iostream> #include <stdio.h> #include <stdlib.h> using namesp

RocketMQ學習筆記（1）----RocketMQ的簡介

1. 什麼是RecketMQ？ 　　 　　 　　是一個佇列模型的訊息中介軟體，具有高效能、高可靠、高實時、分散式特點。　　Producer、Consumer、佇列都可以分散式。　　 Producer 吐一些佇列輪流収送訊息，佇列集合稱為Topic，Consumer 如果做廣播消費，則一個con

RocketMQ學習筆記（4）----RocketMQ搭建雙Master叢集

前面已經學習了RockeMQ的四種叢集方式，接下來就來搭建一個雙Master(2m)的叢集環境。 1. 雙Master伺服器環境 序號　　　　ip　　　　　　使用者名稱　　　　密碼　　　　　　　　角色　　　　　　　　　　　　模式 (1)　　　47.105.145.123　　root　　　　　　　　　　n

RocketMQ學習筆記（5）----RocketMQ監控平臺rocketmq-console-ng的搭建

1. 下載rocketmq-console-ng 　　官網地址：https://github.com/apache/rocketmq-externals 　　 　　拉下來之後，使用idea開啟rocketmq-console，在application.properties配置檔案中修改rocketmq.