顺序消息(FIFO 消息)是阿里云 RocketMQ 提供的一种严格按照顺序来发布和消费的消息类型。本文提供使用 TCP 协议下的开源 Python SDK 收发顺序消息的示例代码供您参考。

顺序消息分为两类:

  • 全局顺序:对于指定的一个 Topic,所有消息按照严格的先入先出(First In First Out,简称 FIFO)的顺序进行发布和消费。
  • 分区顺序:对于指定的一个 Topic,所有消息根据 Sharding Key 进行区块分区。 同一个分区内的消息按照严格的 FIFO 顺序进行发布和消费。Sharding Key 是顺序消息中用来区分不同分区的关键字段,和普通消息的 Key 是完全不同的概念。

详情请参见顺序消息

前提条件

您已完成准备工作。详情请参见准备工作

发送顺序消息

发送顺序消息的示例代码如下。

# -*- coding: utf-8 -*-

from rocketmq.client import Producer, Message

# 发送消息时请设置您在阿里云 RocketMQ 控制台上申请的 Topic。
topic = 'XXXXXX'
# 您在阿里云 RocketMQ 控制台上申请的 GID。
gid = 'GID_XXXXX'
# 设置 TCP 协议接入点,从阿里云 RocketMQ 控制台的实例详情页面获取。
name_srv = 'http://XXXX.aliyuncs.com:80'
# 您在阿里云账号管理控制台中创建的 AccessKeyId,用于身份认证。
ak = 'AK'
# 您在阿里云账号管理控制台中创建的 AccessKeySecret,用于身份认证。
sk = 'SK'
# 用户渠道,默认值为:ALIYUN。
channel = 'ALIYUN'


def create_message():
    # 确保此处设置的 Topic 一定是控制台上申请的顺序类型的。
    msg = Message(topic)
    msg.set_keys('YourKey')
    msg.set_tags('YourTags')
    msg.set_body('Hello RocketMQ, This is a Python Orderly Message.')
    return msg


def send_orderly_with_sharding_key(count):
    producer = Producer(gid, True)
    producer.set_name_server_address(name_srv)
    producer.set_session_credentials(ak, sk, channel)
    producer.start()
    for n in range(count):
        msg = create_message()
        ret = producer.send_orderly_with_sharding_key(msg, 'orderId')
        print ('send message status: ' + str(ret.status) + ' msgId: ' + ret.msg_id)
    print ('send orderly message done')
    producer.shutdown()


if __name__ == '__main__':
    send_orderly_with_sharding_key(10)

消费顺序消息

消费顺序消息的示例代码如下。

# -*- coding: utf-8 -*-

from rocketmq.client import PushConsumer, ConsumeStatus
import time

# 请设置您在阿里云 RocketMQ 控制台上创建的 Topic。
topic = 'XXXXXX'
# 您在阿里云 RocketMQ 控制台上申请的 GID。
gid = 'GID_XXXXX'
# 设置 TCP 协议接入点,从阿里云 RocketMQ 控制台的实例详情页面获取。
name_srv = 'http://XXXX.aliyuncs.com:80'
# 您在阿里云账号管理控制台中创建的 AccessKeyId,用于身份认证。
ak = 'AK'
# 您在阿里云账号管理控制台中创建的 AccessKeySecret,用于身份认证。
sk = 'SK'
# 用户渠道,默认值为:ALIYUN。
channel = 'ALIYUN'


def orderly_callback(msg):
    print(msg.id, msg.body)
    # 消费成功回复 CONSUME_SUCCESS,消费失败回复 RECONSUME_LATER。此时会触发消费重试。
    return ConsumeStatus.CONSUME_SUCCESS


def start_consume_orderly_message():
    # 发送顺序消息必须初始化顺序消息的 Producer,确保第二个参数为 True。
    consumer = PushConsumer(gid, True)
    consumer.set_name_server_address(name_srv)
    consumer.set_session_credentials(ak, sk, channel)
    consumer.subscribe(topic, orderly_callback)
    # ********************************************
    # 1. 确保订阅关系的设置在启动之前完成。
    # 2. 确保相同 GID 下面的消费者的订阅关系一致。
    # 3. 确保此处设置的 Topic 一定是控制台上申请的顺序类型。
    # *********************************************
    print ('start orderly consume message')
    consumer.start()

    # 请保持消费者一直处于运行状态。
    while True:
        time.sleep(3600)


if __name__ == '__main__':
    start_consume_orderly_message()