前提條件

您已完成以下操作：

• 安裝Python SDK。更多信息，請參見準(zhǔn)備環(huán)境。

• 創(chuàng)建資源。代碼中涉及的資源信息，例如實(shí)例、Topic和Group ID等，需要在控制臺上提前創(chuàng)建。更多信息，請參見創(chuàng)建資源。

• 獲取阿里云訪問密鑰AccessKey ID和AccessKey Secret。更多信息，請參見創(chuàng)建AccessKey。

發(fā)送普通消息

發(fā)送普通消息的示例代碼如下。

import sys

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

# 初始化client。
mq_client = MQClient(
    # 設(shè)置HTTP協(xié)議客戶端接入點(diǎn)，進(jìn)入消息隊(duì)列RocketMQ版控制臺實(shí)例詳情頁面的接入點(diǎn)區(qū)域查看。
     "${HTTP_ENDPOINT}",
    # 請確保環(huán)境變量ALIBABA_CLOUD_ACCESS_KEY_ID、ALIBABA_CLOUD_ACCESS_KEY_SECRET已設(shè)置。
    # AccessKey ID，阿里云身份驗(yàn)證標(biāo)識。
    os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
    # AccessKey Secret，阿里云身份驗(yàn)證密鑰。
    os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
    )
# 消息所屬的Topic，在消息隊(duì)列RocketMQ版控制臺創(chuàng)建。
topic_name = "${TOPIC}"
# Topic所屬的實(shí)例ID，在消息隊(duì)列RocketMQ版控制臺創(chuàng)建。
# 若實(shí)例有命名空間，則實(shí)例ID必須傳入；若實(shí)例無命名空間，則實(shí)例ID傳入空字符串。實(shí)例的命名空間可以在消息隊(duì)列RocketMQ版控制臺的實(shí)例詳情頁面查看。
instance_id = "${INSTANCE_ID}"

producer = mq_client.get_producer(instance_id, topic_name)

# 循環(huán)發(fā)送4條消息。
msg_count = 4
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(
                    # 消息內(nèi)容。
                    "I am test message %s.hello" % i,
                    # 消息標(biāo)簽。
                    "tag %s" % i
                        )
            # 設(shè)置消息的自定義屬性。
            msg.put_property("a", i)
            # 設(shè)置消息的Key。
            msg.set_message_key("MessageKey")
            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(
    # 設(shè)置HTTP協(xié)議客戶端接入點(diǎn)，進(jìn)入消息隊(duì)列RocketMQ版控制臺實(shí)例詳情頁面的接入點(diǎn)區(qū)域查看。
    "${HTTP_ENDPOINT}",
    # 請確保環(huán)境變量ALIBABA_CLOUD_ACCESS_KEY_ID、ALIBABA_CLOUD_ACCESS_KEY_SECRET已設(shè)置。
    # AccessKey ID，阿里云身份驗(yàn)證標(biāo)識。
    os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
    # AccessKey Secret，阿里云身份驗(yàn)證密鑰。
    os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
     )
# 消息所屬的Topic，在消息隊(duì)列RocketMQ版控制臺創(chuàng)建。
topic_name = "${TOPIC}"
# 您在消息隊(duì)列RocketMQ版控制臺創(chuàng)建的Group ID。
group_id = "${GROUP_ID}"
# Topic所屬的實(shí)例ID，在消息隊(duì)列RocketMQ版控制臺創(chuàng)建。
# 若實(shí)例有命名空間，則實(shí)例ID必須傳入；若實(shí)例無命名空間，則實(shí)例ID傳入空字符串。實(shí)例的命名空間可以在消息隊(duì)列RocketMQ版控制臺的實(shí)例詳情頁面查看。
instance_id = "${INSTANCE_ID}"

consumer = mq_client.get_consumer(instance_id, topic_name, group_id)

# 長輪詢表示如果Topic沒有消息，則客戶端請求會在服務(wù)端掛起3秒，3秒內(nèi)如果有消息可以消費(fèi)則立即返回響應(yīng)。
# 長輪詢時間3秒（最多可設(shè)置為30秒）。
wait_seconds = 3
# 一次最多消費(fèi)3條（最多可設(shè)置為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:
        # 長輪詢消費(fèi)消息。
        recv_msgs = consumer.consume_message(batch, wait_seconds)
        for msg in recv_msgs:
            print(("Receive, MessageId: %s\nMessageBodyMD5: %s \
                              \nMessageTag: %s\nConsumedTimes: %s \
                              \nPublishTime: %s\nBody: %s \
                              \nNextConsumeTime: %s \
                              \nReceiptHandle: %s \
                              \nProperties: %s\n" % \
                  (msg.message_id, msg.message_body_md5,
                   msg.message_tag, msg.consumed_times,
                   msg.publish_time, msg.message_body,
                   msg.next_consume_time, msg.receipt_handle, msg.properties)))
    except MQExceptionBase as e:
        # Topic中沒有消息可消費(fèi)。
        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前若不確認(rèn)消息消費(fèi)成功，則消息會被重復(fù)消費(fèi)。
    # 消息句柄有時間戳，同一條消息每次消費(fèi)拿到的都不一樣。
    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))
        # 某些消息的句柄可能超時，會導(dǎo)致消息消費(fèi)狀態(tài)確認(rèn)不成功。
        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"])))