rocketmq-python 是一个基于 rocketmq-client-cpp 封装的 RocketMQ Python 客户端。

rocketmq-client-python安装
目前rocketmq库只支持linux和mac。

rocketmq-client-python 的安装：

pip install rocketmq
安装太慢？国内源安装：

pip install rocketmq -i https://pypi.tuna.tsinghua.edu.cn/simple
示例代码：

Producer

from rocketmq.client import Producer, Message
 
producer = Producer('PID-XXX')
producer.set_namesrv_domain('http://onsaddr-internet.aliyun.com/rocketmq/nsaddr4client-internet')#rocketmq队列接口地址（服务器ip:port）
# For ip and port name server address, use `set_namesrv_addr` method, for example:
# producer.set_namesrv_addr('127.0.0.1:9887')
producer.set_session_credentials('XXX', 'XXXX', 'ALIYUN')#可以不使用
producer.start()
 
msg_body = {"id":"test_id","name":"test_name","message":"test_message"}
ss = json.dumps(msg_body).encode('utf-8')
 
msg = Message('YOUR-TOPIC') #topic名称
msg.set_keys('XXX')#每个消息在业务层面的唯一标识码，要设置到keys字段，方便将来定位消息丢失问题。服务器会为每个消息创建索引（哈希索引），应用可以通过topic，key来查询这条消息内容，以及消息被谁消费。由于是哈希索引，请务必保证key尽可能唯一，这样可以避免潜在的哈希冲突。
msg.set_tags('XXX')#一个应用尽可能用一个Topic，消息子类型用tags来标识，tags可以由应用*设置。只有发送消息设置了tags，消费方在订阅消息时，才可以利用tags在broker做消息过滤。
msg.set_body(ss)
ret = producer.send_sync(msg)
print(ret.status, ret.msg_id, ret.offset)
producer.shutdown()

其中：

设置ip:port的位置：producer.set_namesrv_addr('xxx.xxx.xxx.xxx:xxxxx')
当只有单一服务器时，格式是上面这个；

当有多个服务器地址（集群模式）时，可以使用：producer.set_namesrv_addr("xxx.xxx.xxx.xxx:xxxxx,xxx.xxx.xxx.xxx:xxxxx")

如果使用pandas数据，pandas数据可以直接转换
df.to_json(orient='records').encode('utf-8')，然后放入body中发送。

不同应用的多个Topic使用同一个namesrv_addr时数据传输会发生冲突
解决方案：每一个Topic对应一个 “PID-XXX”