消息队列 RocketMQ 5.x版本提供顺序消息(FIFO消息)供您使用。在顺序消息模型中,您需要严格按照顺序来发布和消费消息。本文提供使用PYTHON SDK 收发顺序消息的示例代码供您参考。
顺序消息分为两类,全局顺序消息和分区顺序消息。区别仅为队列数量不同,代码没有区别。
注意顺序消息,需要在控制台申请顺序类型的topic以及顺序类型的group
发送顺序消息的示例代码如下。
from rocketmq import ClientConfiguration, Credentials, Message, Producer 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,)) try: producer.startup() try: msg = Message() msg.topic = topic msg.message_group = "messageGroup" # 顺序topic group 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__': # 火山接入点 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()