博客
关于我
强烈建议你试试无所不能的chatGPT,快点击我
RabbitMQ学习小结(六)—— RPC[Python]
阅读量:3512 次
发布时间:2019-05-20

本文共 3853 字,大约阅读时间需要 12 分钟。

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 tracebackimport pikadef 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 = Nonetry:    # 创建连接    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 pikaimport uuidclass 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. 参考资料

官网资料:

中文资料:

你可能感兴趣的文章
off-by-one
查看>>
ctf-pwn的一些小技巧
查看>>
POJ 1915 Knight Moves
查看>>
Git 撤销修改
查看>>
Git 删除文件
查看>>
Git与远程仓库关联以及关联错误解决方法
查看>>
[HDU] 平方和与立方和
查看>>
[HDU 2096] 小明A+B
查看>>
[HDU 2520] 我是菜鸟,我怕谁(不一样的for循环)
查看>>
[HDU 1215] 七夕节(求因子,不超时)
查看>>
[POJ 1915] Knight Moves
查看>>
Memcache技术精华
查看>>
Redis详解入门篇
查看>>
php开启redis扩展包与redis安装
查看>>
php使用openssl来实现非对称加密
查看>>
pdo如何防止 sql注入
查看>>
myisam和innodb的区别
查看>>
MySQL建表规范与注意事项(个人精华)
查看>>
JDK8接口的新特性
查看>>
synchronized的局限性与lock的好处
查看>>