消息队列 RocketMQ版
消息队列 RocketMQ版
消息队列 RocketMQ版
文档指南
请输入
产品动态
用户指南
实例管理
后台任务管理
Terraform
RocketMQ Agent
API 参考
实例管理
Topic管理
Group管理
消息查询
死信队列
密钥管理
风险巡检
白名单管理
终端节点服务
SDK 参考
RocketMQ 5.x
RocketMQ 4.x
RocketMQ 管控面SDK
最佳实践
- 文档首页
消息队列 RocketMQ版SDK 参考RocketMQ 5.xPython SDK顺序消息
复制全文
下载 pdf顺序消息
文档反馈
问问助手消息队列 RocketMQ 5.x 版本提供顺序消息（FIFO 消息）供您使用。在顺序消息模型中，您需要严格按照顺序来发布和消费消息。本文提供使用 Python SDK 收发顺序消息的示例代码供您参考。
背景信息
- 使用顺序消息前，需在控制台创建顺序消息类型的 Topic 和 Group。
- 顺序消息分为两类，全局顺序消息和分区顺序消息。区别仅为队列数量不同，代码没有区别。
- 全局顺序：
- 对于指定的一个 Topic，所有消息的生产和消费需要遵循一定的顺序，消息的消费顺序必须和生产顺序一致，即需要严格的先入先出 FIFO（First In First Out）的顺序进行发布和消费。
- 分区顺序：
- 对于指定的一个 Topic，其中每一个分区的消息生产与消费是有序的，同一个队列内的消息按照严格的 FIFO 顺序进行发布和订阅。消息投递到哪一个分区由消息的messageGroup来进行区分。
发送顺序消息
发送顺序消息的示例代码如下。
from rocketmq import ClientConfiguration, Credentials, Message, Producer
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,))
try:
producer.startup()
try:
msg = Message()
msg.topic = topic
msg.message_group = "messageGroup" # 顺序消息类型 Topic 的分组。
msg.body = "hello, rocketmq.".encode('utf-8')
msg.tag = "rocketmq-send-fifo-message"
msg.keys = "send_sync"
msg.add_property("send", "sync")
res = producer.send(msg)
print(f"{producer.__str__()} send message success. {res}")
producer.shutdown()
print(f"{producer.__str__()} shutdown.")
except Exception as e:
print(f"normal producer example raise exception: {e}")
producer.shutdown()
except Exception as e:
print(f"{producer.__str__()} startup raise exception: {e}")
producer.shutdown()
订阅顺序消息
订阅顺序消息的示例代码如下。使用前请确认所用 Group 为顺序消息类型。
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:55
这个页面对您有帮助吗？
有用
有用
无用
无用