代码收藏家技术教程 2025-01-01

python如何使用RocketMQ入门

RocketMQ 是一个开源的分布式消息和流数据平台，由阿里巴巴开源。它提供了高性能、低延迟的消息传递服务，并支持多种消息传递模式，包括发布/订阅（Pub/Sub）和点对点（P2P）。

要在 Python 中使用 RocketMQ，可以使用 rocketmq-client-python 这个第三方库。以下是一个简单的入门指南，帮助你在 Python 中使用 RocketMQ。

安装 `rocketmq-client-python`

首先，你需要安装 rocketmq-client-python 库。你可以使用 pip 来安装它：

pip install rocketmq-client-python

RocketMQ 基本概念

在继续之前，了解以下 RocketMQ 的基本概念是有帮助的：

Producer：生产者，负责发送消息到 RocketMQ。

Consumer：消费者，负责从 RocketMQ 接收消息。

Broker：消息代理，负责存储和转发消息。

Topic：主题，消息分类的标识。

Tag：标签，用于进一步区分同一个 Topic 下的不同消息。

发送消息（Producer）

以下是一个简单的示例，展示如何发送消息到 RocketMQ：

from rocketmq.client.producer import Producer, SendResult
from rocketmq.client.exception import MQClientException

# 初始化生产者，指定生产者组名
producer = Producer('example_group_name')

# 设置NameServer地址（替换为你的NameServer地址）
producer.set_namesrv_addr('localhost:9876')

try:
    # 启动生产者
    producer.start()
    for i in range(10):
        # 发送消息
        msg = producer.send_sync('TopicTest', f'Hello RocketMQ {i}'.encode('utf-8'))
        print(f'SendResult: {msg.msg_id}')
finally:
    # 关闭生产者
    producer.shutdown()

接收消息（Consumer）

以下是一个简单的示例，展示如何从 RocketMQ 接收消息：

from rocketmq.client.consumer import PushConsumer, ConsumeConcurrentlyContext, ConsumeConcurrentlyStatus
from rocketmq.client.exception import MQClientException
from rocketmq.common.message import MessageExt

# 初始化消费者，指定消费者组名
consumer = PushConsumer('example_group_name')

# 设置NameServer地址（替换为你的NameServer地址）
consumer.set_namesrv_addr('localhost:9876')

# 订阅主题和标签（'*' 表示订阅该主题下的所有标签）
consumer.subscribe('TopicTest', '*')

# 注册消息监听器
def callback(msgs, context: ConsumeConcurrentlyContext):
    for msg in msgs:
        print(f'Receive message: {msg.body.decode("utf-8")}')
        # 返回消费状态，告诉消费者已成功处理消息
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS

consumer.register_message_listener(callback)

try:
    # 启动消费者
    consumer.start()
    print('Consumer Started.')
except MQClientException as e:
    print(f'MQClientException: {e}')
except KeyboardInterrupt:
    print('Consumer Shutdown.')
finally:
    # 关闭消费者
    consumer.shutdown()