Help Center/ Distributed Message Service for RocketMQ/ Developer Guide/ Python (TCP)/ Sending and Receiving Ordered Messages
Updated on 2024-12-17 GMT+08:00
View PDF
Share

Sending and Receiving Ordered Messages

In DMS for RocketMQ, ordered messages are retrieved in the exact order that they are created.

Ordered messages are ordered globally or on the partition level.

  • Globally ordered messages: There is only one queue in a specific topic. All messages in the queue will be published and subscribed to in the first in, first out (FIFO) order.
  • Partition-level ordered message: Messages within a queue in a specific topic are published and subscribed to in the FIFO order. The producer specifies a partition selection algorithm to ensure that the messages to be ordered are allocated to the same queue.

The only difference between globally ordered messages and partition-level ordered messages is the number of queues. The code is the same.

Before sending and receiving ordered messages, collect RocketMQ connection information by referring to Collecting Connection Information.

Notes and Constraints

To receive and send orderly messages, ensure the topic message type is Orderly before connecting a client to a RocketMQ instance of v5.x.

Sending Ordered Messages

The following code is an example. Replace the information in bold with the actual values.

from rocketmq.client import Producer, Message

topic = 'TopicTest'
gid = 'test'
name_srv = '192.168.0.1:8100'


def create_message():
    msg = Message(topic)
    msg.set_keys('XXX')
    msg.set_tags('XXX')
    msg.set_property('property', 'test')
    msg.set_body('message body')
    return msg


def send_orderly_with_sharding_key():
    producer = Producer(gid, True)
    producer.set_name_server_address(name_srv)
    producer.start()
    msg = create_message()
    ret = producer.send_orderly_with_sharding_key(msg, 'orderId')
    print('send message status: ' + str(ret.status) + ' msgId: ' + ret.msg_id)
    producer.shutdown()


if __name__ == '__main__':
    send_orderly_with_sharding_key()

The parameters in the example code are described as follows. For details about how to obtain the parameter values, see Collecting Connection Information.

  • topic: topic name.
  • gid: producer group name. Enter a name as required.
  • name_srv: instance address and port.

In the preceding code, to ensure the sequence of messages with the same orderId, orderId is used as the sharding key of the specific queue.

Subscribing to Ordered Messages

You only need to add orderly=True to the code for subscribing to normal messages. The following code is an example. Replace the information in bold with the actual values.

import time

from rocketmq.client import PushConsumer, ConsumeStatus


def callback(msg):
    print(msg.id, msg.body, msg.get_property('property'))
    return ConsumeStatus.CONSUME_SUCCESS


def start_consume_message():
    consumer = PushConsumer('consumer_group', orderly=True)
    consumer.set_name_server_address('192.168.0.1:8100')
    consumer.subscribe('TopicTest', callback)
    print('start consume message')
    consumer.start()

    while True:
        time.sleep(3600)


if __name__ == '__main__':
    start_consume_message()

The parameters in the example code are described as follows. For details about how to obtain the parameter values, see Collecting Connection Information.

  • consumer_group: consumer group name.
  • 192.168.0.1:8100: instance address and port.
  • TopicTest: topic name.
Parent topic: Python (TCP)