PythonRabbitMQ消息队列实现rpc-创新互联
上个项目中用到了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()
网页标题:PythonRabbitMQ消息队列实现rpc-创新互联
链接分享:http://ybzwz.com/article/dhcgcj.html