【Python】RocketMQ 基础使用

目录

1. 介绍

2. 实践

2.1. 启动消费者

2.2. 启动生产者

---

1. 介绍

RocketMQ是一个开源的分布式消息传递系统，最初由阿里巴巴集团开发并于2012年开源。它旨在解决高可靠性、高吞吐量、低延迟和可伸缩性等大规模分布式系统下的消息通信需求。

RocketMQ的设计目标是提供一种灵活、可靠、高性能的消息传递解决方案，适用于各种场景，包括在线消息通信、日志处理、流式处理、事件驱动架构等。

下面是对RocketMQ的主要特点和关键功能的简要介绍：

1. 分布式架构：RocketMQ采用了分布式架构，支持水平扩展和高可用性。它的架构包括多个消息生产者、多个消息消费者和多个消息服务器（Broker）。消息生产者将消息发送到Broker，消息消费者从Broker订阅并消费消息。

2. 高可靠性和容错性：RocketMQ通过复制和故障转移机制，提供高可靠性和容错性。它支持主从同步复制和异步复制方式，确保消息不会丢失，并且在Broker故障时能够自动切换到备用Broker。

3. 高吞吐量和低延迟：RocketMQ通过优化存储、网络和消息传递等方面的性能，实现高吞吐量和低延迟。它支持批量发送和批量消费消息，有效地减少网络开销和提高消息处理效率。

4. 灵活的消息模型：RocketMQ支持多种消息模型，包括点对点（P2P）和发布-订阅（Pub-Sub）模型。在P2P模型中，消息生产者直接发送消息给特定的消费者；在Pub-Sub模型中，消息生产者发布消息到特定的主题（Topic），消息消费者订阅感兴趣的主题并接收相应的消息。

5. 丰富的消息过滤和顺序消息支持：RocketMQ提供灵活的消息过滤功能，可以根据消息的属性、标签或SQL表达式进行过滤。此外，它还支持顺序消息，确保相同主题的消息按照发送顺序被消费。

6. 可伸缩性和扩展性：RocketMQ具有良好的可伸缩性和扩展性，可以根据需求增加或减少Broker、生产者和消费者的数量，以适应不断增长的消息流量。

7. 丰富的生态系统和社区支持：RocketMQ拥有活跃的开源社区，提供了丰富的文档、示例和工具。此外，它还与其他开源项目（如Apache Storm、Apache Flume、Apache Flink等）集成，为用户提供更多选择和灵活性。

总之，RocketMQ是一个功能强大的分布式消息传递系统，具有高可靠性、高吞吐量、低延迟和可伸缩性等特点。它适用于构建大规模分布式系统中的消息通信基础设施，为开发者提供了一种可靠、高效的消息传递解决方案。

2. 实践

注意：以下配置需从阿里云 RocketMQ 获取

HTTP_ENDPOINT = "xxx"
ACCESS_KEY = "xxx"
SECRET_KEY = "xxx"

TOPIC = "xxx"
GROUP_ID = "xxx"
INSTANCE_ID = "xxx"

2.1. 启动消费者

""" 
The code was copied from this link
https://help.aliyun.com/zh/apsaramq-for-rocketmq/cloud-message-queue-rocketmq-4-x-series/developer-reference/send-and-subscribe-to-transactional-messages-2?spm=a2c4g.11186623.0.0.4b6f8707cKp9eN
"""
from mq_http_sdk.mq_exception import MQExceptionBase
from mq_http_sdk.mq_consumer import *
from mq_http_sdk.mq_client import *
 
HTTP_ENDPOINT = "xxx"
ACCESS_KEY = "xxx"
SECRET_KEY = "xxx"
 
TOPIC = "xxx"
GROUP_ID = "xxx"
INSTANCE_ID = "xxx"
# 初始化client。
mq_client = MQClient(
    # 设置HTTP协议客户端接入点，进入消息队列RocketMQ版控制台实例详情页面的接入点区域查看。
    HTTP_ENDPOINT,
    # 请确保环境变量ALIBABA_CLOUD_ACCESS_KEY_ID、ALIBABA_CLOUD_ACCESS_KEY_SECRET已设置。
    # AccessKey ID，阿里云身份验证标识。
    ACCESS_KEY,
    # AccessKey Secret，阿里云身份验证密钥。
    SECRET_KEY
     )
# 消息所属的Topic，在消息队列RocketMQ版控制台创建。
topic_name = TOPIC
# 您在消息队列RocketMQ版控制台创建的Group ID。
group_id = GROUP_ID
# Topic所属的实例ID，在消息队列RocketMQ版控制台创建。
# 若实例有命名空间，则实例ID必须传入；若实例无命名空间，则实例ID传入空字符串。实例的命名空间可以在消息队列RocketMQ版控制台的实例详情页面查看。
instance_id = INSTANCE_ID
 
consumer = mq_client.get_consumer(instance_id, topic_name, group_id)
 
# 长轮询表示如果Topic没有消息，则客户端请求会在服务端挂起3秒，3秒内如果有消息可以消费则立即返回响应。
# 长轮询时间3秒（最多可设置为30秒）。
wait_seconds = 3
# 一次最多消费3条（最多可设置为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:
        # 长轮询消费消息。
        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:
        # Topic中没有消息可消费。
        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
 
    # msg.next_consume_time前若不确认消息消费成功，则消息会被重复消费。
    # 消息句柄有时间戳，同一条消息每次消费拿到的都不一样。
    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"])))

2.2. 启动生产者

""" 
The code was copied from this link
https://help.aliyun.com/zh/apsaramq-for-rocketmq/cloud-message-queue-rocketmq-4-x-series/developer-reference/send-and-subscribe-to-transactional-messages-2?spm=a2c4g.11186623.0.0.4b6f8707cKp9eN
"""
import sys
 
from mq_http_sdk.mq_exception import MQExceptionBase
from mq_http_sdk.mq_producer import *
from mq_http_sdk.mq_client import *
import time
import threading
 
HTTP_ENDPOINT = "xxx"
ACCESS_KEY = "xxx"
SECRET_KEY = "xxx"
 
TOPIC = "xxx"
GROUP_ID = "xxx"
INSTANCE_ID = "xxx"
 
# 初始化client。
mq_client = MQClient(
    # 设置HTTP协议客户端接入点，进入消息队列RocketMQ版控制台实例详情页面的接入点区域查看。
    HTTP_ENDPOINT,
    # 请确保环境变量ALIBABA_CLOUD_ACCESS_KEY_ID、ALIBABA_CLOUD_ACCESS_KEY_SECRET已设置。
    # AccessKey ID，阿里云身份验证标识。
    ACCESS_KEY,
    # AccessKey Secret，阿里云身份验证密钥。
    SECRET_KEY
    )
# 消息所属的Topic，在消息队列RocketMQ版控制台创建。
topic_name = TOPIC
# Topic所属的实例ID，在消息队列RocketMQ版控制台创建。
# 若实例有命名空间，则实例ID必须传入；若实例无命名空间，则实例ID传入空字符串。实例的命名空间可以在消息队列RocketMQ版控制台的实例详情页面查看。
instance_id = INSTANCE_ID
 
producer = mq_client.get_producer(instance_id, topic_name)
 
# 循环发送4条消息。
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,
                    # 消息标签。
                    "tag %s" % i
                        )
            # 设置消息的自定义属性。
            msg.put_property("a", i)
            # 设置消息的Key。
            msg.set_message_key("MessageKey")
            re_msg = producer.publish_message(msg)
            print("Publish 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)