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"])))

Funcionamento do long polling

consume_message bloqueia a solicitação no broker por até wait_seconds. Se uma mensagem ficar disponível nesse período, o broker responde imediatamente. Caso contrário, a solicitação retorna após o tempo limite.

Funcionamento da confirmação de mensagens

Após processar uma mensagem, confirme-a enviando o receipt handle de volta ao broker. Se nenhum ACK chegar antes de next_consume_time, o broker reentrega a mensagem com um novo receipt handle. Sempre use o handle da entrega mais recente.