RabbitMQ学习小结(六)—— RPC[Python]
2016-03-02 22:15
726 查看
1. 简介
远程过程调用在很多地方都是经常用到,虽然也有诸多的缺点,但是应用还是较为广泛,对于RPC的概念就不做解释了,本节跟着参考资料,实现基于RabbitMQ的RPC如何使用。接下来我们创建一个模拟RPC服务来返回斐波那契数列。所以需要一个客户端和一个服务端程序。
2. RPC[RabbitMQ]
对于此次操作我们可以理解为客户端对服务端做黑盒操作,客户端不知道服务端的处理逻辑是什么,只知道需要给服务端哪些信息,而服务端从RabbitMQ中取回消息处理完成之后,按照客户端要求将消息发送到指定队列。客户端取到消息后就可以自行处理。基本流程如下图:在上面的描述中,我们可以读取出几个信息,就是客户端要给服务端指定一些信息,而这些信息包含哪些呢?从上图我们可以看到一个reply_to和correlation_id,这两个属性是AMQP中消息的属性,reply_to是我们上面提到的服务端处理完成之后将结果发送给指定的队列;而correlation_id是将RPC的请求和响应关联起来。
AMQP协议给消息预定义了一系列的14个属性。大多数属性很少会用到,除了以下几个:
delivery_mode(投递模式):将消息标记为持久的(值为2)或暂存的(除了2之外的其他任何值)。第二篇教程里接触过这个属性,记得吧?
content_type(内容类型):用来描述编码的mime-type。例如在实际使用中常常使用application/json来描述JOSN编码类型。
reply_to(回复目标):通常用来命名回调队列。
correlation_id(关联标识):用来将RPC的响应和请求关联起来。
如果大家还想对这些属性更加深的了解,可以到官网或者中文资料中查阅,或者点击这里看看AMQP协议的一些简介及属性等内容。
总结
服务端(rpc_server.py)#!/usr/bin/env python # -*- coding: utf-8 -*- # @Date : 2016-03-08 20:38:06 # @Author : mengx (mx472756841@gmail.com) # @Link : http://blog.csdn.net/mx472756841 # @Version : $Id$ ''' 1. 获取请求消息(消费者) 2. 处理请求消息 3. 返回结果消息(生产者) ''' import traceback import pika def fib(n): if n == 0: return 0 elif n == 1: return 1 else: return fib(n-1) + fib(n-2) def on_request(ch, method, props, body): """ 消费处理函数 根据请求的参数,计算斐波那契数列 """ n = int(body) print " [.] fib(%s)" % (n,) # 计算斐波那契 resp = fib(n) # 将结果返回请求指定队列 ch.basic_publish(exchange='', routing_key=props.reply_to, body=str(resp), properties=pika.BasicProperties( correlation_id=props.correlation_id) ) # 消息确认 ch.basic_ack(delivery_tag=method.delivery_tag) conn = None try: # 创建连接 conn = pika.BlockingConnection(pika.ConnectionParameters( host='localhost')) # 创建通道 channel = conn.channel() # 声明队列 channel.queue_declare('rpc_queue') # 在同一时刻处理一条消息 channel.basic_qos(prefetch_count=1) # 消费 channel.basic_consume(on_request, queue='rpc_queue') # 启动 print ' [*] Waiting for messages. To exit press CTRL+C' channel.start_consuming() except: traceback.print_exc() finally: if conn: conn.close()
客户端(rpc_client.py)
#!/usr/bin/env python # -*- coding: utf-8 -*- # @Date : 2016-03-08 20:59:55 # @Author : mengx (mx472756841@gmail.com) # @Link : http://blog.csdn.net/mx472756841 # @Version : $Id$ ''' 1. 请求服务发送消息(作为生产者,指定接受队列) 2. 接收服务返回消息 3. 输出消息 ''' import pika import uuid class FibonacciRpcClient(object): def __init__(self): # 连接 self.connection = pika.BlockingConnection(pika.ConnectionParameters( host='localhost')) # 通道 self.channel = self.connection.channel() # 回调队列 result = self.channel.queue_declare(exclusive=True) self.callback_queue = result.method.queue # 消费处理 self.channel.basic_consume(self.on_response, no_ack=True, queue=self.callback_queue) def on_response(self, ch, method, props, body): # 比较消息是否是自己发送的消息,如果是的话赋值给response if self.corr_id == props.correlation_id: self.response = body def call(self, n): self.response = None self.corr_id = str(uuid.uuid4()) self.channel.basic_publish(exchange='', routing_key='rpc_queue', properties=pika.BasicProperties( reply_to=self.callback_queue, correlation_id=self.corr_id, ), body=str(n)) while self.response is None: # 等待连接的所有消息 self.connection.process_data_events() return int(self.response) fibonacci_rpc = FibonacciRpcClient() print " [x] Requesting fib(30)" response = fibonacci_rpc.call(30) print " [.] Got %r" % (response,)
3. 参考资料
官网资料:http://www.rabbitmq.com/tutorials/tutorial-six-python.html中文资料:RPC/]http://rabbitmq-into-chinese.readthedocs.org/zh_CN/latest/tutorials_with_python/[6]RPC/
相关文章推荐
- RabbitMQ学习小结(四)—— Routing[Python]
- RabbitMQ学习小结(三)—— Publish Subscribe[Python]
- RabbitMQ学习小结(二)—— Work Queues[Python]
- RabbitMQ学习小结(一)—— Hello World [Python]
- python切片解析
- python中的map,filter,zip函数
- python爬虫 bs4_4select()教程
- python爬虫 beutifulsoup4_1官网介绍
- python 使用不同的版本之间的切换
- Python中lambda表达式
- Python的简单爬虫原理及实现
- [大数据]spark入门 in python(一)HelloWorld
- 利用python requests库模拟登陆知乎
- python struct、json、pickle模块
- [leetcode] Insertion Sort List(python)
- 基于Python Django技术构建web系统实践
- python脚本:计算某个目录下Code行数
- python学习小记0
- Python练习(4):牛顿拉复生算法求解根
- 使用Python和Perl绘制北京跑步地图