消息队列 RocketMQ版
消息队列 RocketMQ版
文档指南
请输入
- 文档首页
消息队列 RocketMQ版SDK 参考RocketMQ 5.xPython SDK事务消息
复制全文
下载 pdf事务消息
文档反馈
问问助手本文提供使用 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__':
# 设置为您从火山引擎消息队列 RocketMQ版控制台获取的接入点信息，类似"{INSTANCE_ID}.rocketmq.ivolces.com:8080"。
endpoints = "rocketmq-xxxxx:8080"
# 根据实例支持的认证方式，填写对应的身份认证信息：
# 密钥管理：第一个参数为 AccessKey ID，第二个参数为 AccessKey Secret。
# 权限管理：第一个参数为 ACL 用户名，第二个参数为 ACL 密码。
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__':
# 设置为您从火山引擎消息队列 RocketMQ版控制台获取的接入点信息，类似"{INSTANCE_ID}.rocketmq.ivolces.com:8080"。
endpoints = "rocketmq-xxxxx:8080"
# 根据实例支持的认证方式，填写对应的身份认证信息：
# 密钥管理：第一个参数为 AccessKey ID，第二个参数为 AccessKey Secret。
# 权限管理：第一个参数为 ACL 用户名，第二个参数为 ACL 密码。
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()
最近更新时间：2026.08.26 15:19:56
这个页面对您有帮助吗？
有用
有用
无用
无用