按照rocketmq官网的快速入门,进行linux上的安装:
快速入门 - Apache RocketMQ
避坑:
1、rocketmq可执行命令的路径:
xx/xx/rocketmq/rocketmq-all-4.9.2/distribution/target/rocketmq-4.9.2/rocketmq-4.9.2/bin/
/xx/xx为下载rocketmq时的存放路径
否则会报错:
apache.xxx class noload 依赖的启动类找不到
2、后台启动namesrv和broker
nohup sh bin/mqnamesrv &
nohup sh bin/mqbroker &
使用默认的配置文件启动broker,会导致mq服务只能在服务器本地连接,外部无法连接
producer发消息报错:
rocketmq.exceptions.ProducerSendSyncFailed: No route info of this topic: xxx_topic,error:-1,in file rocketmq-client-cpp/src/producer/DefaultMQProducer.cpp line:379
解决方法:
1、指定配置文件/xx/xx/xx/xx/.../broker_me.conf, 在配置文件中添加以下配置:
namesrvAddr=本地外网IP:9876
brokerIP1=本地外网IP
使用该命令启动:nohup sh bin/mqbroker -c /xx/xx/xx/xx/.../broker_me.conf &
2、服务器开放默认的namesrv端口9876、broker配置文件中的listenPort值(broker端口,默认为10911)
3、日志查看:
namesrv:tail -f ~/logs/rocketmqlogs/namesrv.log
broker:tail -f ~/logs/rocketmqlogs/broker.log
python3使用rocketmq库进行生产和消费rocketmq:
producer:
1 # -*-coding:utf-8-*- 2 from rocketmq.client import Producer, Message 3 4 producer = Producer("producer_id") 5 producer.set_namesrv_addr("namesrvIP:9876") 6 producer.set_group("producer_group") 7 print("start") 8 producer.start() 9 10 def success(sendres): 11 print("success") 12 print(sendres) 13 14 def error(error_obj): 15 print("error") 16 print(error_obj.error) 17 18 msg = Message("xxx_topic") 19 msg.set_body("body_str") 20 # 异步发送 21 producer.send_async(msg, success, error) 22 # 同步发送 23 # ret = producer.send_sync(msg) 24 # print(ret) 25 26 producer.shutdown() 27 print("end...")
consumer:
1 # -*-coding:utf-8-*- 2 import time 3 from rocketmq.client import PushConsumer 4 5 6 def handle_message(msg): 7 print(msg.id) 8 print(msg.body) 9 10 11 consumer = PushConsumer("push_consumer_id") 12 consumer.set_namesrv_addr("namesrvIp:9876") 13 consumer.set_group("push_consumer_group") 14 consumer.subscribe("xxx_topic", handle_message) 15 print("start") 16 consumer.start() 17 18 while True: 19 print("wait 5s") 20 time.sleep(5) 21 22 #consumer.shutdown() 23 #print("end.")