雲訊息佇列 RocketMQ 版提供類似XA或Open XA的分散式交易功能，通過雲訊息佇列 RocketMQ 版事務訊息，能達到分散式交易的最終一致。本文提供使用HTTP協議下的Python SDK收發事務訊息的範例程式碼。
背景資訊
前提條件
您已完成以下操作：
安裝Python SDK。更多資訊，請參見準備環境。
建立資源。代碼中涉及的資源資訊，例如執行個體、Topic和Group ID等，需要在控制台上提前建立。更多資訊，請參見建立資源。
擷取阿里雲存取金鑰AccessKey ID和AccessKey Secret。更多資訊，請參見建立AccessKey。
發送事務訊息
發送事務訊息的範例程式碼如下。
#!/usr/bin/env python
# coding=utf8
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
import threading
# 初始化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}"
# 迴圈發送4條事務訊息。
msg_count = 4
print("%sPublish Transaction Message To %s\nTopicName:%s\nMessageCount:%s\n" \
% (10 * "=", 10 * "=", topic_name, msg_count))
def process_trans_error(exp):
print("\nCommit/Roll Transaction Message Fail! Exception:%s" % exp)
# 如果Commit或Rollback時超過了TransCheckImmunityTime（針對發送事務訊息的控制代碼）或者超過10s（針對consumeHalfMessage的控制代碼），則Commit或Rollback失敗。
if exp.sub_errors:
for sub_error in exp.sub_errors:
print("\tErrorHandle:%s,ErrorCode:%s,ErrorMsg:%s" % \
(sub_error["ReceiptHandle"], sub_error["ErrorCode"], sub_error["ErrorMessage"]))
# 用戶端需要有一個線程或者進程來消費沒有確認的事務訊息。
# 啟動一個線程來檢查沒有確認的事務訊息。
class ConsumeHalfMessageThread(threading.Thread):
def __init__(self):
threading.Thread.__init__(self)
self.count = 0
# 重新New一個Client
self.mq_client = MQClient(
# 設定HTTP接入網域名稱，進入訊息佇列RocketMQ版控制台執行個體詳情頁面的HTTP存取點地區查看。
"${HTTP_ENDPOINT}",
# AccessKey ID阿里雲身分識別驗證，在阿里雲RAM控制台建立。
"${ACCESS_KEY}",
# AccessKey Secret阿里雲身分識別驗證，在阿里雲RAM控制台建立。
"${SECRET_KEY}"
)
self.trans_producer = self.mq_client.get_trans_producer(instance_id, topic_name, group_id)
def run(self):
while 1:
if self.count == 3:
break;
try:
half_msgs = self.trans_producer.consume_half_message(1, 3)
for half_msg in half_msgs:
print("Receive Half Message, MessageId: %s\nMessageBodyMD5: %s \
\nMessageTag: %s\nConsumedTimes: %s \
\nPublishTime: %s\nBody: %s \
\nNextConsumeTime: %s \
\nReceiptHandle: %s \
\nProperties: %s" % \
(half_msg.message_id, half_msg.message_body_md5,
half_msg.message_tag, half_msg.consumed_times,
half_msg.publish_time, half_msg.message_body,
half_msg.next_consume_time, half_msg.receipt_handle, half_msg.properties))
a = int(half_msg.get_property("a"))
try:
if a == 1:
# 確認提交事務訊息。
self.trans_producer.commit(half_msg.receipt_handle)
self.count += 1
print("------>commit")
elif a == 2 and half_msg.consumed_times > 1:
# 確認提交事務訊息。
self.trans_producer.commit(half_msg.receipt_handle)
self.count += 1
print("------>commit")
elif a == 3:
# 確認復原事務訊息。
self.trans_producer.rollback(half_msg.receipt_handle)
self.count += 1
print("------>rollback")
else:
# 什麼都不做，下次再檢查。
print("------>unknown")
except MQExceptionBase as rec_commit_roll_e:
process_trans_error(rec_commit_roll_e)
except MQExceptionBase as half_e:
if half_e.type == "MessageNotExist":
print("No half message! RequestId: %s" % half_e.req_id)
continue
print("Consume half message Fail! Exception:%s\n" % half_e)
break
consume_half_thread = ConsumeHalfMessageThread()
consume_half_thread.setDaemon(True)
consume_half_thread.start()
try:
trans_producer = mq_client.get_trans_producer(instance_id, topic_name, group_id)
for i in range(msg_count):
msg = TopicMessage(
# 訊息內容。
"I am test message %s." % i,
# 訊息標籤。
"tagA"
)
# 設定訊息的自訂屬性。
msg.put_property("a", i)
# 設定訊息的Key。
msg.set_message_key("MessageKey")
# 設定事務第一次回查的時間，為相對時間。單位：秒，範圍：10~300。
# 第一次事務回查後如果訊息沒有Commit或者Rollback，則之後每隔10s左右會回查一次，共回查24小時。
msg.set_trans_check_immunity_time(10)
re_msg = trans_producer.publish_message(msg)
print("Publish Transaction Message Succeed. MessageID:%s, BodyMD5:%s, Handle:%s" \
% (re_msg.message_id, re_msg.message_body_md5, re_msg.receipt_handle))
time.sleep(1)
if i == 0:
# 發送完事務訊息後能擷取到半訊息控制代碼，可以直接Commit或Rollback事務訊息。
try:
trans_producer.commit(re_msg.receipt_handle)
except MQExceptionBase as pub_commit_roll_e:
process_trans_error(pub_commit_roll_e)
except MQExceptionBase as pub_e:
if pub_e.type == "TopicNotExist":
print("Topic not exist, please create it.")
sys.exit(1)
print("Publish Message Fail. Exception:%s" % pub_e)
while 1:
if not consume_half_thread.is_alive():
break
time.sleep(1)訂閱事務訊息
訂閱事務訊息的範例程式碼如下。
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沒有訊息，則用戶端請求會在服務端掛起3秒，3秒內如果有訊息可以消費則立即返迴響應。
# 長輪詢時間3秒（最多可設定為30秒）。
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(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中沒有訊息可消費。
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"])))