黑松山资源网 Design By www.paidiu.com
上个项目中用到了ActiveMQ,只是简单应用,安装完成后直接是用就可以了。由于新项目中一些硬件的限制,需要把消息队列换成RabbitMQ。
RabbitMQ中的几种模式和机制比ActiveMQ多多了,根据业务需要,使用RPC实现功能,其中踩过的一些坑,有必要记录一下了。
上代码,目录结构分为 c_server、c_client、c_hanlder:
c_server:
#!/usr/bin/env python # -*- coding:utf-8 -*- import pika import time import json import io import yaml s_exchange = input("请输入交换机名称-").decode('utf-8').strip() s_queue = input("输入消息队列名称-").decode('utf-8').strip() credentials = pika.PlainCredentials('system', 'manager') connection = pika.BlockingConnection(pika.ConnectionParameters(host='XXX.XXX.XXX.XXX',credentials=credentials)) # 定义 channel = connection.channel() channel.exchange_declare(exchange=s_exchange, exchange_type='direct') channel.queue_declare(queue=s_queue, exclusive=True) channel.queue_bind(queue=s_queue, exchange=s_exchange) def s_manage(content): # 解决unicode转码问题 json.JSONDecoder().decode(content) str_content = yaml.safe_load(json.loads(content,encoding='utf-8')) str_res = { "errorid": 0, "resp": str_content['cmd'], "errorcont": "成功" } return json.dumps(str_res) def on_request(ch, method, props, body): response = s_manage(body) ch.basic_publish(exchange='', routing_key=props.reply_to, properties=pika.BasicProperties(correlation_id = props.correlation_id), body=response) ch.basic_ack(delivery_tag = method.delivery_tag) channel.basic_qos(prefetch_count=1) channel.basic_consume(on_request, queue=s_queue) print(" [x] Awaiting RPC requests") channel.start_consuming()
c_client:
#!/usr/bin/env python # -*- coding:utf-8 -*- import pika import uuid import json import io class RpcClient(object): def __init__(self): self.credentials = pika.PlainCredentials('guest', 'guest') self.connection = pika.BlockingConnection(pika.ConnectionParameters(host='XXX.XXX.XXX.XXX', credentials=self.credentials)) self.channel = self.connection.channel() def on_response(self, ch, method, props, body): if self.callback_id == props.correlation_id: self.response = body ch.basic_ack(delivery_tag=method.delivery_tag) def get_response(self, callback_queue, callback_id): '''取队列里的值,获取callback_queued的执行结果''' self.callback_id = callback_id self.response = None self.channel.queue_declare('q_manager', durable=True) self.channel.basic_consume(self.on_response, # 只要收到消息就执行on_response queue=callback_queue) while self.response is None: self.connection.process_data_events() # 非阻塞版的start_consuming return self.response def call(self, queue_name, command, exchange,rout_key): # 命令下发 '''队列里发送数据''' # result = self.channel.queue_declare(exclusive=False) #exclusive=False 必须这样写 self.callback_queue = 'q_manager' # result.method.queue self.corr_id = str(uuid.uuid4()) self.channel.basic_publish(exchange=exchange, routing_key=queue_name, properties=pika.BasicProperties( reply_to=self.callback_queue, # 发送返回信息的队列name correlation_id=self.corr_id, # 发送uuid 相当于验证码 ), body=command) return self.callback_queue,self.corr_id client
c_handler:
#!/usr/bin/env python # -*- coding:utf-8 -*- from c_client import * import random, time import threading import json import sys class Handler(object): def __init__(self): self.information = {} # 后台进程信息 def check_all(self, *args): '''查看所有信息''' time.sleep(2) print('获取消息') for key in self.information: print("cid【%s】\t 队列【%s】\t 命令【%s】"%(key, self.information[key][0], self.information[key][1])) def check_task(self, cmd): '''查看task_id执行结果''' time.sleep(2) try: task_id = int(cmd) print(task_id) callback_queue= self.information[task_id][2] callback_id= self.information[task_id][3] client = RpcClient() response = client.get_response(callback_queue, callback_id) print(response) # print(response.decode()) del self.information[task_id] except KeyError as e : print("error: [%s]" % e) except IndexError as e: print("error: [%s]" % e) def run(self, user_cmd, host, exchange='', rout_key='',que=''): try: time.sleep(2) command = user_cmd task_id = random.randint(10000, 99999) client = RpcClient() response = client.call(queue_name=host, command=command,exchange=exchange,rout_key=que) self.information[task_id] = [host, command, response[0], response[1]] except IndexError as e: print("[error]:%s"%e) def reflect(self, str,cmd,host,exchange,que): '''反射''' if hasattr(self, str): getattr(self, str)(cmd,host,exchange,que) def start(self, m,cmd, host, exchange,que): while True: user_resp = input("输入处理消息内容ID-").decode('utf-8').strip() self.check_task(user_resp) str = m print(self.information) t1 = threading.Thread(target=self.reflect, args=(str,cmd,host,exchange,que)) #多线程 t1.start() s_exchange = input("请输入交换机名称-").decode('utf-8').strip() s_queue = input("输入消息队列名称-").decode('utf-8').strip() d_cmd_state =input("输入json命令-").decode('utf-8').strip() s_cmd = json.dumps(d_cmd_state) handler = Handler() handler.start('run',s_cmd, s_queue, s_exchange, s_queue) handler
注意要点:1、c_client 发布消息到rabbitmq 需要携带 服务器返回的队列名称,及corr_id
2、c_handler 做了处理,每次发送的内容都会放到task列表中,直到显示ID号,就可以查询返回的内容,调用如下:
以上就是本文的全部内容,希望对大家的学习有所帮助,也希望大家多多支持。
黑松山资源网 Design By www.paidiu.com
广告合作:本站广告合作请联系QQ:858582 申请时备注:广告合作(否则不回)
免责声明:本站资源来自互联网收集,仅供用于学习和交流,请遵循相关法律法规,本站一切资源不代表本站立场,如有侵权、后门、不妥请联系本站删除!
免责声明:本站资源来自互联网收集,仅供用于学习和交流,请遵循相关法律法规,本站一切资源不代表本站立场,如有侵权、后门、不妥请联系本站删除!
黑松山资源网 Design By www.paidiu.com
暂无评论...
稳了!魔兽国服回归的3条重磅消息!官宣时间再确认!
昨天有一位朋友在大神群里分享,自己亚服账号被封号之后居然弹出了国服的封号信息对话框。
这里面让他访问的是一个国服的战网网址,com.cn和后面的zh都非常明白地表明这就是国服战网。
而他在复制这个网址并且进行登录之后,确实是网易的网址,也就是我们熟悉的停服之后国服发布的暴雪游戏产品运营到期开放退款的说明。这是一件比较奇怪的事情,因为以前都没有出现这样的情况,现在突然提示跳转到国服战网的网址,是不是说明了简体中文客户端已经开始进行更新了呢?
更新日志
2024年10月08日
2024年10月08日
- 洪荣宏.2010-上爱的人【华特】【WAV+CUE】
- 《大牌国语情歌 经典再现 深情对唱 2CD》[WAV/分轨][900MB]
- 《陈百强粉红色的一生精选 3CD》[FLAC/分轨][1.2GB]
- 《丁当 20首重量级歌曲 Fu Good 下一站天后 2CD》[WAV/分轨][650MB]
- 歌莉雅.2012-My.Voice.Story【环星】【FLAC分轨】
- 群星.1993-一曲成名·青春无悔【飞碟】【WAV+CUE】
- 刘德华.2000-爱无知(金装版)【NEWMELODY】【WAV+CUE】
- 许巍《时光漫步》1CD[FLAC/分轨][804.3MB]
- 群星《2024第一季度百度歌曲排行抖音榜top100》1CD[FLAC/分轨][2.3GB]
- 张柏芝《全新经验》[FLAC/分轨][450MB]
- 唐朝.2008-浪漫骑士【音乐家】【WAV+CUE】
- 张芸京.2016-失败的高歌【泡耳音乐】【WAV+CUE】
- 群星.1991-音乐工厂1·皇后大道东【音乐工厂】【WAV+CUE】
- 群星.1992-音乐工厂2·首都【音乐工厂】【WAV+CUE】
- 群星.1994-音乐工厂3·儿童乐园【音乐工厂】【WAV+CUE】