python 操作 rocketmq
rocketmq-client-python
RocketMQ Python client, based on rocketmq-client-cpp, supports Linux and macOS
Prerequisites
Install librocketmq
rocketmq-client-python is a lightweight wrapper around rocketmq-client-cpp, so you need install librocketmq first.
Download by binary release.
download specific release according you OS: rocketmq-client-cpp-2.0.0
-
centos
take centos7 as example, you can install the library in centos6 by the same method.
wget https://github.com/apache/rocketmq-client-cpp/releases/download/2.0.0/rocketmq-client-cpp-2.0.0-centos7.x86_64.rpm sudo rpm -ivh rocketmq-client-cpp-2.0.0-centos7.x86_64.rpm -
debian
wget https://github.com/apache/rocketmq-client-cpp/releases/download/2.0.0/rocketmq-client-cpp-2.0.0.amd64.deb sudo dpkg -i rocketmq-client-cpp-2.0.0.amd64.deb -
macOS
wget https://github.com/apache/rocketmq-client-cpp/releases/download/2.0.0/rocketmq-client-cpp-2.0.0-bin-release.darwin.tar.gz tar -xzf rocketmq-client-cpp-2.0.0-bin-release.darwin.tar.gz cd rocketmq-client-cpp mkdir /usr/local/include/rocketmq cp include/* /usr/local/include/rocketmq cp lib/* /usr/local/lib install_name_tool -id "@rpath/librocketmq.dylib" /usr/local/lib/librocketmq.dylib
Build from source
you can also build it manually from source according to Build and Install
Installation
pip install rocketmq-client-pythonUsage
Producer
from rocketmq.client import Producer, Message
producer = Producer('PID-XXX')
producer.set_name_server_address('127.0.0.1:9876')
producer.start()
msg = Message('YOUR-TOPIC')
msg.set_keys('XXX')
msg.set_tags('XXX')
msg.set_body('XXXX')
ret = producer.send_sync(msg)
print(ret.status, ret.msg_id, ret.offset)
producer.shutdown()PushConsumer
import time
from rocketmq.client import PushConsumer, ConsumeStatus
def callback(msg):
print(msg.id, msg.body)
return ConsumeStatus.CONSUME_SUCCESS
consumer = PushConsumer('CID_XXX')
consumer.set_name_server_address('127.0.0.1:9876')
consumer.subscribe('YOUR-TOPIC', callback)
consumer.start()
while True:
time.sleep(3600)
consumer.shutdown()
注意点：
最近需要从rocketmq的队列里读取数据， 显示安装了rocketmq-python跑了几天，后面发现更新里面说代码已经挪位置了（奇怪，我是看到英文的update readme.md才发现…..）,呼~，果断重装了 rocketmq-client-python也是绕了个大圈，相关文档也比较少
rocketmq python库在GitHub地址：https://github.com/apache/rocketmq-client-python
按照页面说明先安装依赖：librocketmq
centos 7：
wget https://github.com/apache/rocketmq-client-cpp/releases/download/2.0.0/rocketmq-client-cpp-2.0.0-centos7.x86_64.rpm
sudo rpm -ivh rocketmq-client-cpp-2.0.0-centos7.x86_64.rpm再安装 rocketmq-client-python
pip install rocketmq-client-python
本以为可以开心的跑起来了，但是问题多多： OSError了，缺少动态链接库
先找到.so 文件的路径
find / -name librocketmq.so
再执行
ln -s /where/you/install/lib/librocketmq.so /usr/libsudo ldconfig
终于跑起来了
#代码参考： 广播 延迟消费 顺序消费 建单消费
-
broadcast
BroadcastConsumer1.py
import time
from rocketmq.client import PushConsumer, ConsumeStatus
# 消息处理回调
from rocketmq.ffi import MessageModel
def callback(msg):
# 模拟业务
print('Received message. messageId: ', msg.id, ' body: ', msg.body)
# 消费成功回复CONSUME_SUCCESS
return ConsumeStatus.CONSUME_SUCCESS
# 消费失败返回RECONSUME_LATER，该消息将会被重新消费
# return ConsumeStatus.RECONSUME_LATER
# 初始化消费者，并设置消费者组信息
consumer = PushConsumer('rocketmq-xxx|namespace_python%group11')
# 设置服务地址
consumer.set_name_server_address('rocketmq-xxx.rocketmq.ap-bj.public.tencenttdmq.com:9876')
# 设置权限（角色名和密钥）
consumer.set_session_credentials(
'eyJrZXlJZC......',
'admin',
''
)
# 设置为集群消息模式
consumer.set_message_model(MessageModel.BROADCASTING)
# 订阅topic
consumer.subscribe('rocketmq-xxx|namespace_python%topic1', callback)
print(' [Consumer] Waiting for messages.')
# 启动消费者
consumer.start()
while True:
time.sleep(3600)
consumer.shutdown()
BroadcastConsumer2
import os
import sys
import time
from rocketmq.client import PushConsumer, ConsumeStatus
# 消息处理回调
from rocketmq.ffi import MessageModel
def callback(msg):
# 模拟业务
print('Received message. messageId: ', msg.id, ' body: ', msg.body)
# 消费成功回复CONSUME_SUCCESS
return ConsumeStatus.CONSUME_SUCCESS
# 消费失败返回RECONSUME_LATER，该消息将会被重新消费
# return ConsumeStatus.RECONSUME_LATER
# 初始化消费者，并设置消费者组信息
consumer = PushConsumer('rocketmq-xxx|namespace_python%group11')
# 设置服务地址
consumer.set_name_server_address('rocketmq-xxx.rocketmq.ap-bj.public.tencenttdmq.com:9876')
# 设置权限（角色名和密钥）
consumer.set_session_credentials(
'eyJrZXlJZC......',
'admin',
''
)
# 设置为集群消息模式
consumer.set_message_model(MessageModel.BROADCASTING)
# 订阅topic
consumer.subscribe('rocketmq-xxx|namespace_python%topic1', callback)
print(' [Consumer] Waiting for messages.')
# 启动消费者
consumer.start()
while True:
time.sleep(3600)
consumer.shutdown()