如何使用TCP協議的JavaSDK收發順序訊息 - ApsaraMQ for RocketMQ

順序訊息（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();
    }
}