消息队列 RocketMQ版
消息队列 RocketMQ版
文档指南
请输入
- 文档首页
消息队列 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
这个页面对您有帮助吗？
有用
有用
无用
无用