本文提供使用HTTP协议下的Python SDK收发定时消息和延时消息的示例代码。
背景信息
延时消息：Producer将消息发送到云消息队列 RocketMQ 版服务端，但并不期望立马投递这条消息，而是延迟一定时间后才投递到Consumer进行消费，该消息即延时消息。
定时消息：Producer将消息发送到云消息队列 RocketMQ 版服务端，但并不期望立马投递这条消息，而是推迟到在当前时间点之后的某一个时间投递到Consumer进行消费，该消息即定时消息。
在HTTP协议下，定时消息和延时消息的代码配置相同，本质都是根据消息中的属性延迟固定时间后才投递给消费者。
更多信息，请参见定时和延时消息。
前提条件
您已完成以下操作：
安装Python SDK。更多信息，请参见准备环境。
创建资源。代码中涉及的资源信息，例如实例、Topic和Group ID等，需要在控制台上提前创建。更多信息，请参见创建资源。
获取阿里云访问密钥AccessKey ID和AccessKey Secret。更多信息，请参见创建AccessKey。
发送定时消息或延时消息
发送定时消息或延时消息的示例代码如下。
import sys
from mq_http_sdk.mq_exception import MQExceptionBase
from mq_http_sdk.mq_producer import *
from mq_http_sdk.mq_client import *
import time
# 初始化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)
# 循环发送4条消息。
msg_count = 4
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,
# 消息标签。
"tag1"
)
# 设置属性。
msg.put_property("a", "i")
# 设置消息的Key。
msg.set_message_key("MessageKey")
# 延时消息，发送时间为10s后。该参数格式为毫秒级别的时间戳。
# 若发送定时消息，设置该参数时需要计算定时时间与当前时间的时间差。
msg.set_start_deliver_time(int(round(time.time() * 1000)) + 10 * 1000)
re_msg = producer.publish_message(msg)
print("Publish Timer 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没有消息，则客户端请求会在服务端挂起3秒，3秒内如果有消息可以消费则立即返回响应。
# 长轮询时间3秒（最多可设置为30秒）。
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(batch, wait_seconds)
for msg in recv_msgs:
print(("Receive, MessageId: %s\nMessageBodyMD5: %s \
\nMessageTag: %s\nConsumedTimes: %s \
\nPublishTime: %s\nBody: %s \
\nNextConsumeTime: %s \
\nReceiptHandle: %s \
\nProperties: %s\n" % \
(msg.message_id, msg.message_body_md5,
msg.message_tag, msg.consumed_times,
msg.publish_time, msg.message_body,
msg.next_consume_time, msg.receipt_handle, msg.properties)))
except MQExceptionBase as e:
# Topic中没有消息可消费。
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"])))文档内容是否对您有帮助？