编程语言
首页 > 编程语言> > 【rocketmq-client-python】学习笔记

【rocketmq-client-python】学习笔记

作者:互联网

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”

标签:set,producer,python,xxx,client,namesrv,msg,rocketmq
来源: https://www.cnblogs.com/java0011/p/15167341.html