全部产品
Search

搜索本产品

    云消息队列 RocketMQ 版：收发事务消息

    文档中心

    云消息队列 RocketMQ 版：收发事务消息

    更新时间：Jul 25, 2023

    云消息队列 RocketMQ 版提供类似XA或Open XA的分布式事务功能，通过云消息队列 RocketMQ 版事务消息，能达到分布式事务的最终一致。本文提供使用HTTP协议下的Python SDK收发事务消息的示例代码。

    背景信息

    事务消息的交互流程如下图所示。

    图片1.png

    更多信息，请参见事务消息。

    前提条件

    您已完成以下操作：

    • 安装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传入空字符串。实例的命名空间可以在消息队列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传入空字符串。实例的命名空间可以在消息队列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"])))