本文提供使用 PYTHON SDK 收发事务消息的示例代码供您参考。

发送事务消息需要在控制台申请事务类型的topic，在发送half消息后，用户可以根据自身业务逻辑对消息选择：

• TransactionImpl.COMMIT：提交事务。事务提交之后，之前暂存的消息对消费者可见。
• TransactionImpl.ROLLBACK：回滚事务。事务回滚，之前暂存的消息会被标记为已经回滚。

另外事务消息需要自定义checker，以便于服收到消息确认后进行回查

from rocketmq import (ClientConfiguration, Credentials, Message, Producer,
                      TransactionChecker, TransactionResolution)

# 未收到若服务端发送者提交的二次确认结果, 服务端进行回查
class TestChecker(TransactionChecker):
    def check(self, message: Message) -> TransactionResolution:
        print(f"do TestChecker check. message_id: {message.message_id}, commit message.")
        return TransactionResolution.COMMIT

if __name__ == '__main__':
    # 火山接入点
    endpoints = "rocketmq-xxxx:8080"
    credentials = Credentials()
    # if auth enable, 火山ak,sk
    credentials = Credentials("xxxx", "xxxx")
    config = ClientConfiguration(endpoints, credentials)
    topic = "xxxx" # 事务topic
    producer = Producer(config, (topic,), TestChecker)

    try:
        producer.startup()
    except Exception as e:
        print(f"{producer.__str__()} startup raise exception: {e}")

    try:
        transaction = producer.begin_transaction()
        msg = Message()
        msg.topic = topic
        msg.body = "hello, rocketmq.".encode('utf-8')
        msg.tag = "rocketmq-send-transaction-message"
        msg.keys = "send_transaction"
        msg.add_property("send", "transaction")
        res = producer.send(msg, transaction)
        print(f"transaction producer{producer.__str__()} send half message success. {res}")
        transaction.commit()
    except Exception as e:
        print(f"transaction producer{producer.__str__()} example raise exception: {e}")

事务消息的订阅方式与普通消息一致，示例代码如下所示。

from rocketmq import ClientConfiguration, Credentials, SimpleConsumer

if __name__ == '__main__':
    # 火山接入点    
    endpoints = "rocketmq-xxxx:8080"
    credentials = Credentials()
    # if auth enable, 火山ak,sk
    credentials = Credentials("xxxx", "xxxx")
    config = ClientConfiguration(endpoints, credentials)
    topic = "xxxx" # topic名称
    simple_consumer = SimpleConsumer(config, "groupName") # group name
    try:
        simple_consumer.startup()
        try:
            simple_consumer.subscribe(topic)
            # use tag filter
            # simple_consumer.subscribe(topic, FilterExpression("tag"))
            while True:
                try:
                    messages = simple_consumer.receive(32, 15)
                    if messages is not None:
                        print(f"{simple_consumer.__str__()} receive {len(messages)} messages.")
                        for msg in messages:
                            simple_consumer.ack(msg)
                            print(f"{simple_consumer.__str__()} ack message:[{msg.message_id}].")
                except Exception as e:
                    print(f"receive or ack message raise exception: {e}")
        except Exception as e:
            print(f"{simple_consumer.__str__()} subscribe topic:{topic} raise exception: {e}")
            simple_consumer.shutdown()
    except Exception as e:
        print(f"{simple_consumer.__str__()} startup raise exception: {e}")
        simple_consumer.shutdown()