火山引擎消息队列 RocketMQ 5.x版本提供同步发送、异步发送两种方式来发送普通消息。本文介绍如何通过不同方式发送普通消息。发送普通消息前请在控制台创建普通消息类型topic

同步发送是指消息发送方发出一条消息后，会在收到服务端返回响应之后才发下一条消息的通讯方式。一般用于较为重要的消息发送场景。
同步发送方式发送普通消息的示例代码如下

from rocketmq import ClientConfiguration, Credentials, Message, Producer

if __name__ == '__main__':
    # 火山接入点
    endpoints = "rocketmq-xxxxx:8080"
    credentials = Credentials()
    # if auth enable, 火山ak,sk
    credentials = Credentials("xxxx", "xxxx")
    config = ClientConfiguration(endpoints, credentials)
    topic = "xxxx" # 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-message"
            msg.keys = "send_sync"
            msg.add_property("send", "sync")
            res = producer.send(msg)
            print(f"{producer.__str__()} send message success. {res}")
            producer.shutdown()
            print(f"{producer.__str__()} shutdown.")
        except Exception as e:
            print(f"normal producer example raise exception: {e}")
            producer.shutdown()
    except Exception as e:
        print(f"{producer.__str__()} startup raise exception: {e}")
        producer.shutdown()

异步发送是指发送方发出一条消息后，不等服务端返回响应，接着发送下一条消息的通讯方式。异步发送可以避免线程阻塞，允许程序继续执行其他任务，从而提高系统的吞吐量和性能。
异步发送方式发送普通消息的示例代码如下

from rocketmq import ClientConfiguration, Credentials, Message, Producer

def handle_send_result(result_future):
    try:
        res = result_future.result()
        print(f"async send message success, {res}")
    except Exception as exception:
        print(f"async send message failed, raise exception: {exception}")


if __name__ == '__main__':
    # 火山接入点    
    endpoints = "rocketmq-xxxx:8080"
    credentials = Credentials()
    # if auth enable, 火山ak,sk
    credentials = Credentials("xxxx", "xxxx")
    config = ClientConfiguration(endpoints, credentials)
    topic = "xxxx"  # 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 = "rocketmq-send-message"
                msg.keys = "send_async"
                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"async producer{producer.__str__()} send message raise exception: {e}")
            producer.shutdown()
    except Exception as e:
        print(f"{producer.__str__()} startup raise exception: {e}")
        producer.shutdown()

5.x RocketMQ支持两种消费模式，分别为 Push Consumer 和 Simple Consumer。前者为服务端推送消息，后者为主动拉取消息，python 5.x SDK仅支持Simple Consumer
Simple Consumer消费示例代码如下：

from rocketmq import ClientConfiguration, Credentials, SimpleConsumer

if __name__ == '__main__':
    # 火山接入点    
    endpoints = "rocketmq-xxxx:8080"
    credentials = Credentials()
    # if auth enable, 火山ak,sk
    credentials = Credentials("xxxx", "xxxx")
    config = ClientConfiguration(endpoints, credentials)
    topic = "xxxx" # topic名称
    simple_consumer = SimpleConsumer(config, "groupName") # group name
    try:
        simple_consumer.startup()
        try:
            simple_consumer.subscribe(topic)
            # use tag filter
            # simple_consumer.subscribe(topic, FilterExpression("tag"))
            while True:
                try:
                    messages = simple_consumer.receive(32, 15)
                    if messages is not None:
                        print(f"{simple_consumer.__str__()} receive {len(messages)} messages.")
                        for msg in messages:
                            simple_consumer.ack(msg)
                            print(f"{simple_consumer.__str__()} ack message:[{msg.message_id}].")
                except Exception as e:
                    print(f"receive or ack message raise exception: {e}")
        except Exception as e:
            print(f"{simple_consumer.__str__()} subscribe topic:{topic} raise exception: {e}")
            simple_consumer.shutdown()
    except Exception as e:
        print(f"{simple_consumer.__str__()} startup raise exception: {e}")
        simple_consumer.shutdown()