ApsaraMQ for RocketMQ:Envío y recepción de mensajes programados y diferidos

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

Funcionamiento del sondeo largo

consume_message bloquea la solicitud en el broker durante un máximo de wait_seconds. Si un mensaje está disponible durante ese periodo, el broker responde de inmediato. De lo contrario, la solicitud finaliza al cumplirse el tiempo de espera.

Funcionamiento de la confirmación de mensajes

Después de procesar un mensaje, confírmelo al enviar el identificador de recibo de vuelta al broker. Si no llega ninguna confirmación antes de next_consume_time, el broker vuelve a entregar el mensaje con un nuevo identificador de recibo. Use siempre el identificador de la entrega más reciente.