#! /usr/bin/python
#encoding:utf-8
from rocketmq.client import Producer, Message
import json
import sys
import time
reload(sys)
sys.setdefaultencoding( "utf-8" )
'''
rocketmq写入消息
'''
def send():

 producer = Producer('test')
 producer.set_namesrv_addr('127.0.0.1:9876')  #rocketmq队列接口地址（服务器ip:port）
 producer.start()

 msg_body = {"id":"001","name":"test_mq","message":"abcdefg"}
 ss = json.dumps(msg_body).encode('utf-8')

 msg = Message('test')   #topic名称
 msg.set_keys('ce')
 msg.set_tags('ce')
 msg.set_body(ss)      #message body

 retmq = producer.send_sync(msg)
 print(retmq.status, retmq.msg_id, retmq.offset)
 producer.shutdown()


if __name__ =="__main__":
    print "开始时间：" + time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
    for i in range(1000):
     send()
    print "结束时间：" + time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))

#消费方式PullConsumer（全部消费）（可重复消费）
from rocketmq.client import PullConsumer
import json
consumer = PullConsumer('PID-001')
consumer.set_namesrv_addr('ip:port')
consumer.start()
for msg in consumer.pull('test'):
    print(msg.id, msg.body)
consumer.shutdown()

#! /usr/bin/python
#encoding:utf-8
from rocketmq.client import Producer, Message
from rocketmq.client import PullConsumer,PushConsumer

import json
import sys
import time
reload(sys)
sys.setdefaultencoding( "utf-8" )
#消费方式PushConsumer不可重复消费
def callback(msg):
    print(msg)
print "开始时间：" + time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))
consumer = PushConsumer('test')
consumer.set_namesrv_addr('127.0.0.1:9876')

consumer.subscribe("test",callback)
consumer.start()

while True:
  time.sleep(30)

consumer.shutdown() 
print "结束时间：" + time.strftime('%Y-%m-%d %H:%M:%S', time.localtime(time.time()))