全部產品
Search
文件中心

ApsaraMQ for RocketMQ:收發順序訊息

更新時間:Jul 01, 2024

順序訊息(FIFO訊息)是雲訊息佇列 RocketMQ 版提供的一種嚴格按照順序來發布和消費的訊息類型。本文提供使用TCP協議下的Java SDK收發順序訊息的範例程式碼。

前提條件

您已完成以下操作:

  • 下載1.2.7或以上版本的Java SDK。更多資訊,請參見Java SDK版本說明

  • 準備環境。更多資訊,請參見準備環境

  • (可選)日誌配置。更多資訊,請參見日誌配置

背景資訊

順序訊息分為兩類:

  • 全域順序:對於指定的一個Topic,所有訊息按照嚴格的先入先出FIFO(First In First Out)的順序進行發布和消費。

  • 分區順序:對於指定的一個Topic,所有訊息根據Sharding Key進行區塊分區。同一個分區內的訊息按照嚴格的FIFO順序進行發布和消費。Sharding Key是順序訊息中用來區分不同分區的關鍵字段,和普通訊息的Key是完全不同的概念。

更多資訊,請參見順序訊息

說明

對於新手使用者,建議在正式收發訊息前,閱讀Demo工程來瞭解搭建雲訊息佇列 RocketMQ 版工程的具體步驟。

發送順序訊息

重要

雲訊息佇列 RocketMQ 版服務端判定訊息產生的順序性是參照單一生產者、單一線程並發下訊息發送的時序。如果發送方有多個生產者或者有多個線程並發發送訊息,則此時只能以到達雲訊息佇列 RocketMQ 版服務端的時序作為訊息順序的依據,和業務側的發送順序未必一致。

具體的範例程式碼,請以雲訊息佇列 RocketMQ 版程式碼程式庫為準。

全域順序訊息和分區順序訊息的發布方式基本一樣,範例程式碼如下。

package com.aliyun.openservices.ons.example.order;

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.SendResult;
import com.aliyun.openservices.ons.api.order.OrderProducer;

import java.util.Properties;


public class ProducerClient {

    public static void main(String[] args) {
        Properties properties = new Properties();
        // 您在訊息佇列RocketMQ版控制台建立的Group ID。
        properties.put(PropertyKeyConst.GROUP_ID,"XXX");
        // AccessKey ID,阿里雲身分識別驗證標識。
        properties.put(PropertyKeyConst.AccessKey,"XXX");
        // AccessKey Secret,阿里雲身分識別驗證密鑰。
        properties.put(PropertyKeyConst.SecretKey,"XXX");
        // 設定TCP接入網域名稱,進入訊息佇列RocketMQ版控制台執行個體詳情頁面的存取點地區查看。
        properties.put(PropertyKeyConst.NAMESRV_ADDR,"XXX");
        OrderProducer producer = ONSFactory.createOrderProducer(properties);
        // 在發送訊息前,必須調用start方法來啟動Producer,只需調用一次即可。
        producer.start();
        for (int i = 0; i < 1000; i++) {
            String orderId = "biz_" + i % 10;
            Message msg = new Message(
                    // Message所屬的Topic。
                    "Order_global_topic",
                    // Message Tag,可理解為Gmail中的標籤,對訊息進行再歸類,方便Consumer指定過濾條件在訊息佇列RocketMQ版的伺服器過濾。
                    "TagA",
                    // Message Body,可以是任何二進位形式的資料,訊息佇列RocketMQ版不做任何幹預,需要Producer與Consumer協商好一致的序列化和還原序列化方式。
                    "send order global msg".getBytes()
            );
            // 設定代表訊息的業務關鍵屬性,請儘可能全域唯一。
            // 以方便您在無法正常收到訊息情況下,可通過訊息佇列RocketMQ版控制台查詢訊息並補發。
            // 注意:不設定也不會影響訊息正常收發。
            msg.setKey(orderId);
            // 分區順序訊息中區分不同分區的關鍵字段,Sharding Key與普通訊息的key是完全不同的概念。
            // 全域順序訊息,該欄位可以設定為任意非Null 字元串。
            String shardingKey = String.valueOf(orderId);
            try {
                SendResult sendResult = producer.send(msg, shardingKey);
                // 發送訊息,只要不拋異常就是成功。
                if (sendResult != null) {
                    System.out.println(new Date() + " Send mq message success. Topic is:" + msg.getTopic() + " msgId is: " + sendResult.getMessageId());
                }
            }
            catch (Exception e) {
                // 訊息發送失敗,需要進行重試處理,可重新發送這條訊息或持久化這條資料進行補償處理。
                System.out.println(new Date() + " Send mq message failed. Topic is:" + msg.getTopic());
                e.printStackTrace();
            }
        }
        // 在應用退出前,銷毀Producer對象。
        // 注意:銷毀Producer對象可以節約系統記憶體,若您需要頻繁發送訊息,則無需銷毀Producer對象。
        producer.shutdown();
    }

}

訂閱順序訊息

全域順序訊息和分區順序訊息的訂閱者式基本一樣,範例程式碼如下。

package com.aliyun.openservices.ons.example.order;

import com.aliyun.openservices.ons.api.Message;
import com.aliyun.openservices.ons.api.ONSFactory;
import com.aliyun.openservices.ons.api.PropertyKeyConst;
import com.aliyun.openservices.ons.api.order.ConsumeOrderContext;
import com.aliyun.openservices.ons.api.order.MessageOrderListener;
import com.aliyun.openservices.ons.api.order.OrderAction;
import com.aliyun.openservices.ons.api.order.OrderConsumer;

import java.util.Properties;


public class ConsumerClient {

    public static void main(String[] args) {
        Properties properties = new Properties();
        // 您在訊息佇列RocketMQ版控制台建立的Group ID。
        properties.put(PropertyKeyConst.GROUP_ID,"XXX");
        // AccessKey ID,阿里雲身分識別驗證標識。
        properties.put(PropertyKeyConst.AccessKey,"XXX");
        // AccessKey Secret,阿里雲身分識別驗證密鑰。
        properties.put(PropertyKeyConst.SecretKey,"XXX");
        // 設定TCP接入網域名稱,進入訊息佇列RocketMQ版控制台執行個體詳情頁面的存取點地區查看。
        properties.put(PropertyKeyConst.NAMESRV_ADDR,"XXX");
          // 順序訊息消費失敗進行重試前的等待時間,單位(毫秒),取值範圍: 10毫秒~30,000毫秒。
        properties.put(PropertyKeyConst.SuspendTimeMillis,"100");
        // 訊息消費失敗時的最大重試次數。
        properties.put(PropertyKeyConst.MaxReconsumeTimes,"20");

        // 在訂閱訊息前,必須調用start方法來啟動Consumer,只需調用一次即可。
        OrderConsumer consumer = ONSFactory.createOrderedConsumer(properties);

        consumer.subscribe(
                // Message所屬的Topic。
                "Order_global_topic",
                // 訂閱指定Topic下的Tags:
                // 1. * 表示訂閱所有訊息。
                // 2. TagA || TagB || TagC表示訂閱TagA或TagB或TagC的訊息。
                "*",
                new MessageOrderListener() {
                    /**
                     * 1. 訊息消費處理失敗或者處理出現異常,返回OrderAction.Suspend。
                     * 2. 訊息處理成功,返回OrderAction.Success。
                     */
                    @Override
                    public OrderAction consume(Message message, ConsumeOrderContext context) {
                        System.out.println(message);
                        return OrderAction.Success;
                    }
                });

        consumer.start();
    }
}