python rocketmq生产者消费者

安装依赖包

pip install rocketmq

生产者
需要注意的是假如你用的java SDK 需要只是UNinname
我们可以看到下列代码设置了tag以及key，在页面可以根据key查找消息


from rocketmq.client import Producer, Message
import jsonproducer = Producer('PID-XXX')
# producer.set_namesrv_domain('http://onsaddr-internet.aliyun.com/rocketmq/nsaddr4client-internet')
# For ip and port name server address, use `set_namesrv_addr` method, for example:
producer.set_namesrv_addr('192.168.214.134:9876')
producer.set_max_message_size(1024*1024)# producer.set_session_credentials('XXX', 'XXXX', 'ALIYUN') # No need to call this function if you don't use Aliyun.
producer.start()msg = Message('ellis')
msg.set_keys('XXX')
msg.set_tags('XXX')
msg.set_body(json.dumps({"key1":"value1"}))
ret = producer.send_sync(msg)print(ret.msg_id,ret.offset,ret.status)
producer.shutdown()

消费方式PullConsumer（全部消费）（可重复消费）


```python
from rocketmq.client import PullConsumerconsumer = PullConsumer('CID_XXX')
consumer.set_namesrv_domain('http://onsaddr-internet.aliyun.com/rocketmq/nsaddr4client-internet')
# For ip and port name server address, use `set_namesrv_addr` method, for example:
# consumer.set_namesrv_addr('127.0.0.1:9887')
consumer.set_session_credentials('XXX', 'XXXX', 'ALIYUN') # No need to call this function if you don't use Aliyun.
consumer.start()for msg in consumer.pull('YOUR-TOPIC'):print(msg.id, msg.body)consumer.shutdown()

消费方式PushConsumer（即时消费）（不可重复消费）

import timefrom rocketmq.client import PushConsumerdef callback(msg):print(msg.id, msg.body.decode())consumer = PushConsumer('ellis1')
consumer.set_group("ellis1")
consumer.set_instance_name("ellis1")
# consumer.set_namesrv_domain('http://onsaddr-internet.aliyun.com/rocketmq/nsaddr4client-internet')
# For ip and port name server address, use `set_namesrv_addr` method, for example:
consumer.set_namesrv_addr('127.0.0.1:9876')
# consumer.set_session_credentials('XXX', 'XXXX', 'ALIYUN') # No need to call this function if you don't use Aliyun.
其中tags是要订阅的TAG
consumer.subscribe('ellis', callback,TAGS)
consumer.start()while True:time.sleep(3600)consumer.shutdown()

生产者发送消息选择队列，以及设置顺序

from rocketmq.client import Producer, Message
import jsondef queue_selector(mq_size, msg, arg):return 1producer = Producer('PID-XXX')
# producer.set_namesrv_domain('http://onsaddr-internet.aliyun.com/rocketmq/nsaddr4client-internet')
# For ip and port name server address, use `set_namesrv_addr` method, for example:
producer.set_namesrv_addr('192.168.214.134:9876')
producer.set_max_message_size(1024*1024)# producer.set_session_credentials('XXX', 'XXXX', 'ALIYUN') # No need to call this function if you don't use Aliyun.
producer.start()msg = Message('ellis')
msg.set_keys('XXX')
msg.set_tags('XXX')
msg.set_body(json.dumps({"key1":"value1"}))ret = producer.send_oneway_orderly(msg,arg=1,queue_selector=queue_selector)producer.shutdown()

消费者顺序消费

import timefrom rocketmq.client import PushConsumerdef callback(msg):print(msg)print(msg.id, msg.body.decode())consumer = PushConsumer('ellis1',orderly=True)
consumer.set_group("ellis3")
consumer.set_instance_name("ellis1")
# consumer.set_namesrv_domain('http://onsaddr-internet.aliyun.com/rocketmq/nsaddr4client-internet')
# For ip and port name server address, use `set_namesrv_addr` method, for example:
consumer.set_namesrv_addr('127.0.0.1:9876')
# consumer.set_session_credentials('XXX', 'XXXX', 'ALIYUN') # No need to call this function if you don't use Aliyun.
consumer.subscribe('ellis', callback)
consumer.start()while True:time.sleep(3600)consumer.shutdown()