Sample code - ApsaraMQ for RocketMQ: Central de Documentação do Alibaba Cloud

Pré-requisitos

Antes de começar, verifique se você tem:

O SDK do RocketMQ para Python instalado. Para detalhes sobre versões, consulte Guia de versões
Uma instância do ApsaraMQ for RocketMQ 5.x com um endpoint. Para instruções de configuração, consulte Preparação do ambiente
Um tópico e um grupo de consumidores criados na instância

Parâmetros

Substitua os seguintes marcadores de posição em cada exemplo de código pelos seus valores reais.

Parâmetro	Valor de exemplo	Descrição
`<your-endpoint>`	`rmq-cn-xxx.{regionId}.rmq.aliyuncs.com:8080`	Endpoint da sua instância do ApsaraMQ for RocketMQ. Para obter o endpoint, consulte Obter o endpoint da instância. Use o endpoint público para acesso pela internet ou o endpoint de VPC para acesso via VPC.
`<your-instance-id>`	`rmq-cn-xxx`	ID da sua instância do ApsaraMQ for RocketMQ. Necessário apenas para acesso pela rede pública em instâncias serverless.
`<your-topic>`	`normal_test`	Tópico de envio ou consumo de mensagens. Crie o tópico antecipadamente.
`<your-consumer-group>`	`GID_test`	Grupo de consumidores usado para consumir mensagens. Crie o grupo de consumidores antecipadamente.
`<your-ak>`	`1XVg0hzgKm******`	Nome de usuário da sua instância. Obrigatório para acesso pela internet. Para acesso via VPC, é necessário apenas se a instância for serverless e a autenticação livre em VPCs estiver desabilitada. Consulte Obter o nome de usuário e a senha da instância.
`<your-sk>`	`ijSt8rEc45******`	Senha da sua instância. Segue as mesmas condições de acesso do nome de usuário.

Enviar mensagens normais

As mensagens normais não possuem semântica especial de entrega e representam o tipo mais comum de mensagem.

Envio síncrono

O método send() bloqueia a execução até que o broker confirme o recebimento da mensagem.

from rocketmq import ClientConfiguration, Credentials, Message, Producer

if __name__ == '__main__':
    # Instance endpoint
    endpoints = "<your-endpoint>"
    # Authenticate with your instance username and password
    credentials = Credentials("<your-ak>", "<your-sk>")
    config = ClientConfiguration(endpoints, credentials)
    topic = "<your-topic>"
    producer = Producer(config, (topic,))

    try:
        producer.startup()
        try:
            msg = Message()
            msg.topic = topic
            msg.body = "hello, rocketmq.".encode('utf-8')
            # Tag: secondary classifier of messages besides topic
            msg.tag = "tag"
            # Keys: alternative way to identify messages besides message ID
            msg.keys = "keys"
            # Custom user property
            msg.add_property("key", "value")
            for i in range(10):
                res = producer.send(msg)
                print(f"Send message success. {res}")
            producer.shutdown()
        except Exception as e:
            print(f"Send message failed: {e}")
            producer.shutdown()
    except Exception as e:
        print(f"Producer startup failed: {e}")
        producer.shutdown()

Envio assíncrono

O método send_async() retorna imediatamente e invoca um callback quando o broker responde. Use o envio assíncrono para obter maior throughput quando não for necessário aguardar cada confirmação individualmente.

from rocketmq import ClientConfiguration, Credentials, Message, Producer

def handle_send_result(result_future):
    try:
        # Avoid time-consuming logic in the callback; offload to another thread if needed
        res = result_future.result()
        print(f"Send message success. {res}")
    except Exception as exception:
        print(f"Send message failed: {exception}")

if __name__ == '__main__':
    endpoints = "<your-endpoint>"
    credentials = Credentials("<your-ak>", "<your-sk>")
    config = ClientConfiguration(endpoints, credentials)
    topic = "<your-topic>"
    producer = Producer(config, (topic,))

    try:
        producer.startup()
        try:
            for i in range(10):
                msg = Message()
                msg.topic = topic
                msg.body = "hello, rocketmq.".encode('utf-8')
                msg.tag = "tag"
                msg.keys = "keys"
                msg.add_property("send", "async")
                send_result_future = producer.send_async(msg)
                send_result_future.add_done_callback(handle_send_result)
        except Exception as e:
            print(f"Send message failed: {e}")
    except Exception as e:
        print(f"Producer startup failed: {e}")

    input("Press Enter to stop the application.")
    producer.shutdown()

Enviar mensagens ordenadas

As mensagens ordenadas são entregues na mesma ordem de envio dentro de um mesmo grupo de mensagens. Defina a propriedade message_group para agrupar mensagens relacionadas.

from rocketmq import ClientConfiguration, Credentials, Message, Producer

if __name__ == '__main__':
    endpoints = "<your-endpoint>"
    credentials = Credentials("<your-ak>", "<your-sk>")
    config = ClientConfiguration(endpoints, credentials)
    topic = "<your-topic>"
    producer = Producer(config, (topic,))

    try:
        producer.startup()
        try:
            msg = Message()
            msg.topic = topic
            msg.body = "hello, rocketmq.".encode('utf-8')
            msg.tag = "rocketmq-send-fifo-message"
            msg.keys = "keys"
            # Messages with the same message_group are delivered in order
            msg.message_group = "fifo-group"
            for i in range(10):
                res = producer.send(msg)
                print(f"Send message success. {res}")
            producer.shutdown()
        except Exception as e:
            print(f"Send message failed: {e}")
            producer.shutdown()
    except Exception as e:
        print(f"Producer startup failed: {e}")
        producer.shutdown()

Nota

O source para mensagens ordenadas está disponível em fifo_producer_example.py.

Enviar mensagens agendadas e atrasadas

As mensagens agendadas e atrasadas são entregues após um timestamp especificado. Defina delivery_timestamp com um horário Unix epoch em segundos.

import time

from rocketmq import ClientConfiguration, Credentials, Message, Producer

if __name__ == '__main__':
    endpoints = "<your-endpoint>"
    credentials = Credentials("<your-ak>", "<your-sk>")
    config = ClientConfiguration(endpoints, credentials)
    topic = "<your-topic>"
    producer = Producer(config, (topic,))

    try:
        producer.startup()
        try:
            msg = Message()
            msg.topic = topic
            msg.body = "hello, rocketmq.".encode('utf-8')
            msg.tag = "rocketmq-send-delay-message"
            # Deliver the message 10 seconds from now
            msg.delivery_timestamp = int(time.time()) + 10
            res = producer.send(msg)
            print(f"Send message success. {res}")
            producer.shutdown()
        except Exception as e:
            print(f"Send message failed: {e}")
            producer.shutdown()
    except Exception as e:
        print(f"Producer startup failed: {e}")
        producer.shutdown()

Enviar mensagens transacionais

As mensagens transacionais suportam commit em duas fases. O broker consulta o producer para confirmar ou reverter mensagens pendentes por meio de um TransactionChecker.

from rocketmq import (ClientConfiguration, Credentials, Message, Producer,
                      TransactionChecker, TransactionResolution)

class OrderTransactionChecker(TransactionChecker):
    """Check whether the local transaction succeeded and return COMMIT or ROLLBACK."""

    def check(self, message: Message) -> TransactionResolution:
        print(f"Transaction check for message: {message}. Committing.")
        return TransactionResolution.COMMIT

if __name__ == '__main__':
    endpoints = "<your-endpoint>"
    credentials = Credentials("<your-ak>", "<your-sk>")
    config = ClientConfiguration(endpoints, credentials)
    topic = "<your-topic>"
    # Pass the transaction checker to the producer
    producer = Producer(config, (topic,), checker=OrderTransactionChecker())

    try:
        producer.startup()
    except Exception as e:
        print(f"Producer startup failed: {e}")

    try:
        # Begin a half-message transaction
        transaction = producer.begin_transaction()
        msg = Message()
        msg.topic = topic
        msg.body = "hello, rocketmq.".encode('utf-8')
        msg.tag = "rocketmq-send-transaction-message"
        res = producer.send(msg, transaction)
        print(f"Send half message success. {res}")

        # Option 1: Commit from client side
        transaction.commit()
        print(f"Committed message: {transaction.message_id}")

        # Option 2: Roll back instead
        # transaction.rollback()
        # print(f"Rolled back message: {transaction.message_id}")

        producer.shutdown()
    except Exception as e:
        print(f"Transaction send failed: {e}")
        producer.shutdown()

Enviar mensagens lite

As mensagens lite usam um tópico pai com sub-tópicos (lite topics) para categorizar mensagens com granularidade mais fina. Defina lite_topic na mensagem para especificar o sub-tópico.

from rocketmq import ClientConfiguration, Credentials, Message, Producer

if __name__ == '__main__':
    endpoints = "<your-endpoint>"
    credentials = Credentials("<your-ak>", "<your-sk>")
    config = ClientConfiguration(endpoints, credentials)
    # Use the parent topic
    topic = "<your-topic>"
    producer = Producer(config, (topic,))

    try:
        producer.startup()
        try:
            msg = Message()
            msg.topic = topic
            msg.body = "hello, rocketmq.".encode('utf-8')
            for i in range(10):
                # Route each message to a lite topic under the parent topic
                msg.lite_topic = f"lite-test-{i}"
                res = producer.send(msg)
                print(f"Send message success. {res}")
            producer.shutdown()
        except Exception as e:
            print(f"Send message failed: {e}")
            producer.shutdown()
    except Exception as e:
        print(f"Producer startup failed: {e}")
        producer.shutdown()

Consumir mensagens com um simple consumer

Um simple consumer busca mensagens do broker usando long polling. Após processar cada mensagem, chame ack() para confirmá-la.

from rocketmq import (ClientConfiguration, Credentials, FilterExpression,
                      SimpleConsumer)

if __name__ == '__main__':
    endpoints = "<your-endpoint>"
    credentials = Credentials("<your-ak>", "<your-sk>")
    config = ClientConfiguration(endpoints, credentials)
    topic = "<your-topic>"
    consumer_group = "<your-consumer-group>"
    # Use a singleton consumer instance in production; avoid creating multiple consumers
    simple_consumer = SimpleConsumer(config, consumer_group, {topic: FilterExpression()})

    try:
        simple_consumer.startup()
        try:
            # Optional: filter by tag
            # simple_consumer.subscribe(topic, FilterExpression("your-tag"))
            while True:
                try:
                    # Long poll for up to 32 messages; each message is invisible
                    # for 15 seconds after receipt
                    messages = simple_consumer.receive(32, 15)
                    if messages is not None:
                        for msg in messages:
                            simple_consumer.ack(msg)
                            print(f"Acknowledged message: [{msg.message_id}]")
                except Exception as e:
                    print(f"Receive or ack failed: {e}")
        except Exception as e:
            print(f"Consumer error: {e}")
            simple_consumer.shutdown()
    except Exception as e:
        print(f"Consumer startup failed: {e}")
        simple_consumer.shutdown()

Consumir mensagens lite com um push consumer

Um push consumer recebe mensagens automaticamente por meio de um listener registrado. Para ver o exemplo de push consumer para mensagens lite, consulte lite_push_consumer_example.py.

Acesso pela rede pública para instâncias serverless

Para acessar uma instância serverless do ApsaraMQ for RocketMQ pela rede pública, passe o ID da instância como terceiro argumento em ClientConfiguration:

config = ClientConfiguration(endpoints, credentials, "<your-instance-id>")

Substitua <your-instance-id> pelo ID real da sua instância, por exemplo rmq-cn-xxx.

Arquivos source no GitHub

A tabela a seguir associa cada tipo de mensagem ao seu arquivo source correspondente no GitHub.

Tipo de mensagem	Envio	Consumo
Mensagens normais	Síncrono: normal_producer_example.py, Assíncrono: async_producer_example.py	simple_consumer_example.py
Mensagens ordenadas	fifo_producer_example.py	-
Mensagens agendadas e atrasadas	delay_producer_example.py	-
Mensagens transacionais	transaction_producer_example.py	-
Mensagens lite	lite_producer_example.py	lite_push_consumer_example.py

ApsaraMQ for RocketMQ:Sample code