# 收发事务消息
分布式消息服务RocketMQ版的事务消息支持在业务逻辑与发送消息之间提供事务保证，通过两阶段的方式提供对事务消息的支持，事务消息交互流程如[图1]所示。
图1事务消息交互流程   
![](https://support.huaweicloud.com/devg-hrm/zh-cn_image_0000001129284708.png "点击放大")
事务消息生产者首先发送半消息，然后执行本地事务。如果执行成功，则发送事务提交，否则发送事务回滚。服务端在一段时间后如果一直收不到提交或回滚，则发起回查，生产者在收到回查后重新发送事务提交或回滚。消息只有在提交之后才投递给消费者，消费者对回滚的消息不可见。
收发事务消息前，请参考[收集连接信息](https://support.huaweicloud.com/devg-hrm/hrm-devg-008.html)收集RocketMQ所需的连接信息。
#### 约束与限制
RocketMQ TCP协议的Python SDK只支持Linux系统，且不支持开启SSL。
#### 发送事务消息
参考如下示例代码（以下加粗内容需要替换为实例自有信息，请根据实际情况替换）。
```
import time
from rocketmq.client import Message, TransactionMQProducer, TransactionStatus
topic = 'TopicTest'
name_srv = '192.168.0.1:8100'
def create_message():
    msg = Message(topic)
    msg.set_keys('messageKey')
    msg.set_tags('messageTag')
    msg.set_property('property', 'test')
    msg.set_body('message body')
    return msg
def check_callback(msg):
    print('check: ' + msg.body.decode('utf-8'))
    return TransactionStatus.COMMIT
def local_execute(msg, user_args):
    print('local:   ' + msg.body.decode('utf-8'))
    return TransactionStatus.UNKNOWN
def send_transaction_message(count):
    producer = TransactionMQProducer('producerGroup', check_callback)
    producer.set_name_server_address(name_srv)
    producer.start()
    for n in range(count):
        msg = create_message()
        ret = producer.send_message_in_transaction(msg, local_execute, None)
        print('send message status: ' + str(ret.status) + ' msgId: ' + ret.msg_id)
    print('send transaction message done')
    while True:
        time.sleep(3600)
if __name__ == '__main__':
    send_transaction_message(10)
```
示例代码中的参数说明如下，请参考[收集连接信息](https://support.huaweicloud.com/devg-hrm/hrm-devg-008.html)获取参数值。
- topic：表示Topic名称。
- name_srv：表示实例连接地址和端口。
- set_keys：设置消息索引键，可根据关键字精确查找某条消息。
- set_tags：设置消息Tag，用于消费端根据指定Tag过滤消息。
事务消息生产者需要实现两个回调函数，其中local_execute回调函数在发送完半事务消息后被调用，即上图中的第3阶段，check_callback回调函数在收到回查时调用，即上图中的第6阶段。两个回调函数均可返回3种事务状态：
- TransactionStatus.COMMIT：提交事务，允许消费者消费该消息。
- TransactionStatus.ROLLBACK：回滚事务，消息将被丢弃不允许消费。
- TransactionStatus.UNKNOWN：无法判断状态，期待服务端向生产者再次回查该消息的状态。
#### 订阅事务消息
订阅事务消息的代码与[订阅普通消息的代码](https://support.huaweicloud.com/devg-hrm/hrm-devg-018.html#hrm-devg-018__section77006348111)相同。