全部产品
Search
文档中心

云消息队列 RocketMQ 版:收发顺序消息

更新时间:Jul 25, 2023

顺序消息(FIFO消息)是云消息队列 RocketMQ 版提供的一种严格按照顺序来发布和消费的消息类型。本文提供使用HTTP协议下的Python SDK收发顺序消息的示例代码。

背景信息

顺序消息分为两类:

  • 全局顺序:对于指定的一个Topic,所有消息按照严格的先入先出FIFO(First In First Out)的顺序进行发布和消费。

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

更多信息,请参见顺序消息

前提条件

您已完成以下操作:

  • 安装Python SDK。更多信息,请参见准备环境

  • 创建资源。代码中涉及的资源信息,例如实例、Topic和Group ID等,需要在控制台上提前创建。更多信息,请参见创建资源

  • 获取阿里云访问密钥AccessKey ID和AccessKey Secret。更多信息,请参见创建AccessKey

发送顺序消息

重要

云消息队列 RocketMQ 版服务端判定消息产生的顺序性是参照单一生产者、单一线程并发下消息发送的时序。如果发送方有多个生产者或者有多个线程并发发送消息,则此时只能以到达云消息队列 RocketMQ 版服务端的时序作为消息顺序的依据,和业务侧的发送顺序未必一致。

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

import sys

from mq_http_sdk.mq_exception import MQExceptionBase
from mq_http_sdk.mq_producer import *
from mq_http_sdk.mq_client import *

# 初始化client。
mq_client = MQClient(
    # 设置HTTP协议客户端接入点,进入消息队列RocketMQ版控制台实例详情页面的接入点区域查看。
     "${HTTP_ENDPOINT}",
    # 请确保环境变量ALIBABA_CLOUD_ACCESS_KEY_ID、ALIBABA_CLOUD_ACCESS_KEY_SECRET已设置。
    # AccessKey ID,阿里云身份验证标识。
    os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
    # AccessKey Secret,阿里云身份验证密钥。
    os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
    )
# 消息所属的Topic,在消息队列RocketMQ版控制台创建。
topic_name = "${TOPIC}"
# Topic所属的实例ID,在消息队列RocketMQ版控制台创建。
# 若实例有命名空间,则实例ID必须传入;若实例无命名空间,则实例ID传入空字符串。实例的命名空间可以在消息队列RocketMQ版控制台的实例详情页面查看。
instance_id = "${INSTANCE_ID}"

producer = mq_client.get_producer(instance_id, topic_name)

# 循环发送8条消息。
msg_count = 8
print("%sPublish Message To %s\nTopicName:%s\nMessageCount:%s\n" % (10 * "=", 10 * "=", topic_name, msg_count))

try:
    for i in range(msg_count):
        msg = TopicMessage(
            # 消息内容。
            "I am test message %s.hello" % i, 
            # 消息标签。
            "tag %s" % i
        )
        # 设置消息的自定义属性。
        msg.put_property("a", str(i))
        # 设置分区顺序消息的Sharding Key,用于标识不同的分区。Sharding Key与消息的Key是完全不同的概念。
        msg.set_sharding_key(str(i % 3))
        re_msg = producer.publish_message(msg)
        print("Publish Message Succeed. MessageID:%s, BodyMD5:%s" % (re_msg.message_id, re_msg.message_body_md5))
except MQExceptionBase as e:
    if e.type == "TopicNotExist":
        print("Topic not exist, please create it.")
        sys.exit(1)
    print("Publish Message Fail. Exception:%s" % e)
            

订阅顺序消息

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

from mq_http_sdk.mq_exception import MQExceptionBase
from mq_http_sdk.mq_consumer import *
from mq_http_sdk.mq_client import *

# 初始化client。
mq_client = MQClient(
    # 设置HTTP协议客户端接入点,进入消息队列RocketMQ版控制台实例详情页面的接入点区域查看。
     "${HTTP_ENDPOINT}",
    # 请确保环境变量ALIBABA_CLOUD_ACCESS_KEY_ID、ALIBABA_CLOUD_ACCESS_KEY_SECRET已设置。
    # AccessKey ID,阿里云身份验证标识。
    os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
    # AccessKey Secret,阿里云身份验证密钥。
    os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
    )
# 消息所属的Topic,在消息队列RocketMQ版控制台创建。
topic_name = "${TOPIC}"
# 您在消息队列RocketMQ版控制台创建的Group ID。
group_id = "${GROUP_ID}"
# Topic所属的实例ID,在消息队列RocketMQ版控制台创建。
# 若实例有命名空间,则实例ID必须传入;若实例无命名空间,则实例ID传入空字符串。实例的命名空间可以在消息队列RocketMQ版控制台的实例详情页面查看。
instance_id = "${INSTANCE_ID}"

consumer = mq_client.get_consumer(instance_id, topic_name, group_id)
# 长轮询顺序消费消息, 拉取到的消息可能是多个分区的(对于分区顺序消息),一个分区内的消息一定是顺序的。
# 对于分区顺序消息,只要一个分区内存在没有被确认消费的消息,那么该分区下次还会消费到相同的消息。
# 对于一个分区,只有所有消息确认消费成功才能消费下一批消息。
# 长轮询表示如果Topic没有消息,则请求会在服务端挂起3s,3s内如果有消息可以消费则服务端立即返回响应。
wait_seconds = 3
# 一次最多消费3条(最多可设置为16条)。
batch = 3
print(("%sConsume And Ak Message From Topic%s\nTopicName:%s\nMQConsumer:%s\nWaitSeconds:%s\n" \
        % (10 * "=", 10 * "=", topic_name, group_id, wait_seconds)))
while True:
    try:
        recv_msgs = consumer.consume_message_orderly(batch, wait_seconds)
        print("=======>Receive %d messages:" % len(recv_msgs))
        for msg in recv_msgs:
            print("\tMessageId: %s, MessageBodyMD5: %s,NextConsumeTime: %s,ConsumedTimes: %s,PublishTime: %s\n\tBody: %s \
                    \n\tReceiptHandle: %s \
                    \n\tProperties: %s,ShardingKey: %s\n" % \
                  (msg.message_id, msg.message_body_md5,
                   msg.next_consume_time, msg.consumed_times,
                   msg.publish_time, msg.message_body,
                   msg.receipt_handle, msg.properties, msg.get_sharding_key()))
    except MQExceptionBase as e:
        if e.type == "MessageNotExist":
            print(("No new message! RequestId: %s" % e.req_id))
            continue

        print(("Consume Message Fail! Exception:%s\n" % e))
        time.sleep(2)
        continue

    # msg.next_consume_time前若不确认消息消费成功,则消息会被重复消费。
    # 消息句柄有时间戳,同一条消息每次消费拿到的都不一样。
    try:
        receipt_handle_list = [msg.receipt_handle for msg in recv_msgs]
        consumer.ack_message(receipt_handle_list)
        print(("========>Ak %s Message Succeed.\n\n" % len(receipt_handle_list)))
    except MQExceptionBase as e:
        print(("\nAk Message Fail! Exception:%s" % e))
        # 某些消息的句柄可能超时,会导致消息消费状态确认不成功。
        if e.sub_errors:
            for sub_error in e.sub_errors:
                print(("\tErrorHandle:%s,ErrorCode:%s,ErrorMsg:%s" % \
                      (sub_error["ReceiptHandle"], sub_error["ErrorCode"], sub_error["ErrorMessage"])))