收發順序訊息 - ApsaraMQ for RocketMQ

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

背景資訊

順序訊息分為兩類：

全域順序：對於指定的一個Topic，所有訊息按照嚴格的先入先出FIFO（First In First Out）的順序進行發布和消費。
分區順序：對於指定的一個Topic，所有訊息根據Sharding Key進行區塊分區。同一個分區內的訊息按照嚴格的FIFO順序進行發布和消費。Sharding Key是順序訊息中用來區分不同分區的關鍵字段，和普通訊息的Key是完全不同的概念。

更多資訊，請參見順序訊息。

前提條件

您已完成以下操作：

安裝Java SDK。更多資訊，請參見準備環境。
建立資源。代碼中涉及的資源資訊，例如執行個體、Topic和Group ID等，需要在控制台上提前建立。更多資訊，請參見建立資源。
擷取阿里雲存取金鑰AccessKey ID和AccessKey Secret。更多資訊，請參見建立AccessKey。

發送順序訊息

重要

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

發送順序訊息的範例程式碼如下。

import com.aliyun.mq.http.MQClient;
import com.aliyun.mq.http.MQProducer;
import com.aliyun.mq.http.model.TopicMessage;

import java.util.Date;

public class OrderProducer {

    public static void main(String[] args) {
        MQClient mqClient = new MQClient(
                // 設定HTTP協議用戶端存取點，進入訊息佇列RocketMQ版控制台執行個體詳情頁面的存取點地區查看。
                "${HTTP_ENDPOINT}",
                // 請確保環境變數ALIBABA_CLOUD_ACCESS_KEY_ID、ALIBABA_CLOUD_ACCESS_KEY_SECRET已設定。
                // AccessKey ID，阿里雲身分識別驗證標識。
	              System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"),
	              // AccessKey Secret，阿里雲身分識別驗證密鑰。
	              System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
        );

        // 訊息所屬的Topic，在訊息佇列RocketMQ版控制台建立。
        // 不同訊息類型的Topic不能混用，例如普通訊息的Topic只能用於收發普通訊息，不能用於收發其他類型的訊息。
        final String topic = "${TOPIC}";
        // Topic所屬的執行個體ID，在訊息佇列RocketMQ版控制台建立。
        // 若執行個體有命名空間，則執行個體ID必須傳入；若執行個體無命名空間，則執行個體ID傳入null空值或字串空值。執行個體的命名空間可以在訊息佇列RocketMQ版控制台的執行個體詳情頁面查看。
        final String instanceId = "${INSTANCE_ID}";

        // 擷取Topic的生產者。
        MQProducer producer;
        if (instanceId != null && instanceId != "") {
            producer = mqClient.getProducer(instanceId, topic);
        } else {
            producer = mqClient.getProducer(topic);
        }

        try {
            // 迴圈發送8條訊息。
            for (int i = 0; i < 8; i++) {
                TopicMessage pubMsg = new TopicMessage(
                        // 訊息內容。
                        "hello mq!".getBytes(),
                        // 訊息標籤。
                        "A"
                );
                // 設定分區順序訊息的Sharding Key，用於標識不同的分區。Sharding Key與訊息的Key是完全不同的概念。
                pubMsg.setShardingKey(String.valueOf(i % 2));
                pubMsg.getProperties().put("a", String.valueOf(i));
                // 同步發送訊息，只要不拋異常就是成功。
                TopicMessage pubResultMsg = producer.publishMessage(pubMsg);

                // 同步發送訊息，只要不拋異常就是成功。
                System.out.println(new Date() + " Send mq message success. Topic is:" + topic + ", msgId is: " + pubResultMsg.getMessageId()
                        + ", bodyMD5 is: " + pubResultMsg.getMessageBodyMD5());
            }
        } catch (Throwable e) {
            // 訊息發送失敗，需要進行重試處理，可重新發送這條訊息或持久化這條資料進行補償處理。
            System.out.println(new Date() + " Send mq message failed. Topic is:" + topic);
            e.printStackTrace();
        }

        mqClient.close();
    }

}

訂閱順序訊息

訂閱順序訊息的範例程式碼如下。

import com.aliyun.mq.http.MQClient;
import com.aliyun.mq.http.MQConsumer;
import com.aliyun.mq.http.common.AckMessageException;
import com.aliyun.mq.http.model.Message;

import java.util.ArrayList;
import java.util.List;

public class OrderConsumer {

    public static void main(String[] args) {
        MQClient mqClient = new MQClient(
                // 設定HTTP協議用戶端存取點，進入訊息佇列RocketMQ版控制台執行個體詳情頁面的存取點地區查看。
                "${HTTP_ENDPOINT}",
                // 請確保環境變數ALIBABA_CLOUD_ACCESS_KEY_ID、ALIBABA_CLOUD_ACCESS_KEY_SECRET已設定。
                // AccessKey ID，阿里雲身分識別驗證標識。
	              System.getenv("ALIBABA_CLOUD_ACCESS_KEY_ID"),
	              // AccessKey Secret，阿里雲身分識別驗證密鑰。
	              System.getenv("ALIBABA_CLOUD_ACCESS_KEY_SECRET")
        );

        // 訊息所屬的Topic，在訊息佇列RocketMQ版控制台建立。
        //不同訊息類型的Topic不能混用，例如普通訊息的Topic只能用於收發普通訊息，不能用於收發其他類型的訊息。
        final String topic = "${TOPIC}";
        // 您在訊息佇列RocketMQ版控制台建立的Group ID。
        final String groupId = "${GROUP_ID}";
        // Topic所屬的執行個體ID，在訊息佇列RocketMQ版控制台建立。
        // 若執行個體有命名空間，則執行個體ID必須傳入；若執行個體無命名空間，則執行個體ID傳入null空值或字串空值。執行個體的命名空間可以在訊息佇列RocketMQ版控制台的執行個體詳情頁面查看。
        final String instanceId = "${INSTANCE_ID}";

        final MQConsumer consumer;
        if (instanceId != null && instanceId != "") {
            consumer = mqClient.getConsumer(instanceId, topic, groupId, null);
        } else {
            consumer = mqClient.getConsumer(topic, groupId);
        }

        // 在當前線程迴圈消費訊息，建議多開個幾個線程並發消費訊息。
        do {
            List<Message> messages = null;

            try {
                // 長輪詢順序消費訊息, 拉取到的訊息可能是多個分區的（對於分區順序訊息），一個分區內的訊息一定是順序的。
                // 對於分區順序訊息，只要一個分區記憶體在沒有被確認消費的訊息，那麼該分區下次還會消費到相同的訊息。
                // 對於一個分區，只有所有訊息確認消費成功才能消費下一批訊息。
                // 長輪詢表示如果Topic沒有訊息，則請求會在服務端掛起3s，3s內如果有訊息可以消費則服務端立即返迴響應。
                messages = consumer.consumeMessageOrderly(
                        3,  // 一次最多消費3條訊息（最多可設定為16條）。
                        3   // 長輪詢時間為3秒（最多可設定為30秒）。
                );
            } catch (Throwable e) {
                e.printStackTrace();
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e1) {
                    e1.printStackTrace();
                }
            }
            // Topic中沒有訊息可消費。
            if (messages == null || messages.isEmpty()) {
                System.out.println(Thread.currentThread().getName() + ": no new message, continue!");
                continue;
            }

            // 處理商務邏輯。
            System.out.println("Receive " + messages.size() + " messages:");
            for (Message message : messages) {
                System.out.println(message);
                System.out.println("ShardingKey: " + message.getShardingKey() + ", a:" + message.getProperties().get("a"));
            }

            // 訊息重試時間到達前若不確認訊息消費成功，則訊息會被重複消費。
            // 訊息控制代碼有時間戳記，同一條訊息每次消費拿到的都不一樣。
            {
                List<String> handles = new ArrayList<String>();
                for (Message message : messages) {
                    handles.add(message.getReceiptHandle());
                }

                try {
                    consumer.ackMessage(handles);
                } catch (Throwable e) {
                    // 某些訊息的控制代碼可能逾時，會導致訊息消費狀態確認不成功。
                    if (e instanceof AckMessageException) {
                        AckMessageException errors = (AckMessageException) e;
                        System.out.println("Ack message fail, requestId is:" + errors.getRequestId() + ", fail handles:");
                        if (errors.getErrorMessages() != null) {
                            for (String errorHandle :errors.getErrorMessages().keySet()) {
                                System.out.println("Handle:" + errorHandle + ", ErrorCode:" + errors.getErrorMessages().get(errorHandle).getErrorCode()
                                        + ", ErrorMsg:" + errors.getErrorMessages().get(errorHandle).getErrorMessage());
                            }
                        }
                        continue;
                    }
                    e.printStackTrace();
                }
            }
        } while (true);
    }
}