收發順序訊息 - ApsaraMQ for RocketMQ

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

背景資訊

順序訊息分為兩類：

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

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

前提條件

您已完成以下操作：

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

發送順序訊息

重要

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

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

import sys

from mq_http_sdk.mq_exception import MQExceptionBase
from mq_http_sdk.mq_producer import *
from mq_http_sdk.mq_client import *

# 初始化client。
mq_client = MQClient(
    # 設定HTTP協議用戶端存取點，進入訊息佇列RocketMQ版控制台執行個體詳情頁面的存取點地區查看。
     "${HTTP_ENDPOINT}",
    # 請確保環境變數ALIBABA_CLOUD_ACCESS_KEY_ID、ALIBABA_CLOUD_ACCESS_KEY_SECRET已設定。
    # AccessKey ID，阿里雲身分識別驗證標識。
    os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
    # AccessKey Secret，阿里雲身分識別驗證密鑰。
    os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
    )
# 訊息所屬的Topic，在訊息佇列RocketMQ版控制台建立。
topic_name = "${TOPIC}"
# Topic所屬的執行個體ID，在訊息佇列RocketMQ版控制台建立。
# 若執行個體有命名空間，則執行個體ID必須傳入；若執行個體無命名空間，則執行個體ID傳入Null 字元串。執行個體的命名空間可以在訊息佇列RocketMQ版控制台的執行個體詳情頁面查看。
instance_id = "${INSTANCE_ID}"

producer = mq_client.get_producer(instance_id, topic_name)

# 迴圈發送8條訊息。
msg_count = 8
print("%sPublish Message To %s\nTopicName:%s\nMessageCount:%s\n" % (10 * "=", 10 * "=", topic_name, msg_count))

try:
    for i in range(msg_count):
        msg = TopicMessage(
            # 訊息內容。
            "I am test message %s.hello" % i, 
            # 訊息標籤。
            "tag %s" % i
        )
        # 設定訊息的自訂屬性。
        msg.put_property("a", str(i))
        # 設定分區順序訊息的Sharding Key，用於標識不同的分區。Sharding Key與訊息的Key是完全不同的概念。
        msg.set_sharding_key(str(i % 3))
        re_msg = producer.publish_message(msg)
        print("Publish Message Succeed. MessageID:%s, BodyMD5:%s" % (re_msg.message_id, re_msg.message_body_md5))
except MQExceptionBase as e:
    if e.type == "TopicNotExist":
        print("Topic not exist, please create it.")
        sys.exit(1)
    print("Publish Message Fail. Exception:%s" % e)

訂閱順序訊息

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

from mq_http_sdk.mq_exception import MQExceptionBase
from mq_http_sdk.mq_consumer import *
from mq_http_sdk.mq_client import *

# 初始化client。
mq_client = MQClient(
    # 設定HTTP協議用戶端存取點，進入訊息佇列RocketMQ版控制台執行個體詳情頁面的存取點地區查看。
     "${HTTP_ENDPOINT}",
    # 請確保環境變數ALIBABA_CLOUD_ACCESS_KEY_ID、ALIBABA_CLOUD_ACCESS_KEY_SECRET已設定。
    # AccessKey ID，阿里雲身分識別驗證標識。
    os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
    # AccessKey Secret，阿里雲身分識別驗證密鑰。
    os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
    )
# 訊息所屬的Topic，在訊息佇列RocketMQ版控制台建立。
topic_name = "${TOPIC}"
# 您在訊息佇列RocketMQ版控制台建立的Group ID。
group_id = "${GROUP_ID}"
# Topic所屬的執行個體ID，在訊息佇列RocketMQ版控制台建立。
# 若執行個體有命名空間，則執行個體ID必須傳入；若執行個體無命名空間，則執行個體ID傳入Null 字元串。執行個體的命名空間可以在訊息佇列RocketMQ版控制台的執行個體詳情頁面查看。
instance_id = "${INSTANCE_ID}"

consumer = mq_client.get_consumer(instance_id, topic_name, group_id)
# 長輪詢順序消費訊息, 拉取到的訊息可能是多個分區的（對於分區順序訊息），一個分區內的訊息一定是順序的。
# 對於分區順序訊息，只要一個分區記憶體在沒有被確認消費的訊息，那麼該分區下次還會消費到相同的訊息。
# 對於一個分區，只有所有訊息確認消費成功才能消費下一批訊息。
# 長輪詢表示如果Topic沒有訊息，則請求會在服務端掛起3s，3s內如果有訊息可以消費則服務端立即返迴響應。
wait_seconds = 3
# 一次最多消費3條（最多可設定為16條）。
batch = 3
print(("%sConsume And Ak Message From Topic%s\nTopicName:%s\nMQConsumer:%s\nWaitSeconds:%s\n" \
        % (10 * "=", 10 * "=", topic_name, group_id, wait_seconds)))
while True:
    try:
        recv_msgs = consumer.consume_message_orderly(batch, wait_seconds)
        print("=======>Receive %d messages:" % len(recv_msgs))
        for msg in recv_msgs:
            print("\tMessageId: %s, MessageBodyMD5: %s,NextConsumeTime: %s,ConsumedTimes: %s,PublishTime: %s\n\tBody: %s \
                    \n\tReceiptHandle: %s \
                    \n\tProperties: %s,ShardingKey: %s\n" % \
                  (msg.message_id, msg.message_body_md5,
                   msg.next_consume_time, msg.consumed_times,
                   msg.publish_time, msg.message_body,
                   msg.receipt_handle, msg.properties, msg.get_sharding_key()))
    except MQExceptionBase as e:
        if e.type == "MessageNotExist":
            print(("No new message! RequestId: %s" % e.req_id))
            continue

        print(("Consume Message Fail! Exception:%s\n" % e))
        time.sleep(2)
        continue

    # msg.next_consume_time前若不確認訊息消費成功，則訊息會被重複消費。
    # 訊息控制代碼有時間戳記，同一條訊息每次消費拿到的都不一樣。
    try:
        receipt_handle_list = [msg.receipt_handle for msg in recv_msgs]
        consumer.ack_message(receipt_handle_list)
        print(("========>Ak %s Message Succeed.\n\n" % len(receipt_handle_list)))
    except MQExceptionBase as e:
        print(("\nAk Message Fail! Exception:%s" % e))
        # 某些訊息的控制代碼可能逾時，會導致訊息消費狀態確認不成功。
        if e.sub_errors:
            for sub_error in e.sub_errors:
                print(("\tErrorHandle:%s,ErrorCode:%s,ErrorMsg:%s" % \
                      (sub_error["ReceiptHandle"], sub_error["ErrorCode"], sub_error["ErrorMessage"])))