ApsaraMQ for RocketMQ:Send and receive scheduled and delayed messages

import sys
import os
import time

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

# Initialize the producer client
mq_client = MQClient(
    # HTTP endpoint. Find it in the HTTP Endpoint section
    # of the Instance Details page in the ApsaraMQ for RocketMQ console.
    "<http-endpoint>",
    # AccessKey ID and AccessKey secret for authentication.
    # Store credentials in environment variables instead of hardcoding them.
    os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
    os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
)

topic_name = "<topic>"
instance_id = "<instance-id>"

producer = mq_client.get_producer(instance_id, topic_name)

# Send four messages
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,  # Message content
            "tag1"                               # Message tag
        )
        # Set a custom message attribute
        msg.put_property("a", "i")
        # Set the message key for tracing
        msg.set_message_key("MessageKey")

        # Deliver this message after a 10-second delay.
        # The value is a Unix timestamp in milliseconds.
        # For a scheduled message, replace the calculation below
        # with the target delivery time in milliseconds.
        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)

import os
import time

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

# Initialize the consumer client
mq_client = MQClient(
    "<http-endpoint>",
    os.environ['ALIBABA_CLOUD_ACCESS_KEY_ID'],
    os.environ['ALIBABA_CLOUD_ACCESS_KEY_SECRET']
)

topic_name = "<topic>"
group_id = "<group-id>"
instance_id = "<instance-id>"

consumer = mq_client.get_consumer(instance_id, topic_name, group_id)

# Long polling timeout in seconds (max: 30)
wait_seconds = 3
# Maximum messages per request (max: 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:
        # Long polling: the request blocks on the broker until a message
        # arrives or wait_seconds elapses, whichever comes first.
        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:
        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

    # Acknowledge consumed messages.
    # If the broker does not receive an ACK before msg.next_consume_time,
    # it redelivers the message. Each delivery generates a new receipt handle.
    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"])))

Uzun anketlemenin çalışma mantığı

consume_message çağrısı, aracıda en fazla wait_seconds süre boyunca bloke olur. Bu süre içinde bir mesaj kullanılabilir hale gelirse aracı hemen yanıt verir. Aksi takdirde istek zaman aşımından sonra döner.

Mesaj onaylamanın çalışma mantığı

Bir mesajı işledikten sonra makbuz tanıtıcısını (receipt handle) aracıya geri göndererek onaylayın. next_consume_time öncesinde bir ACK alınmazsa aracı mesajı yeni bir makbuz tanıtıcısıyla yeniden teslim eder. Her zaman en son teslimattan gelen tanıtıcıyı kullanın.