RabbitMQ消息队列-RPC

RPC

远程过程调用（英语：Remote Procedure Call，缩写为 RPC）是一个计算机通信协议。该协议允许运行于一台计算机的程序调用另一台计算机的子程序，而程序员无需额外地为这个交互作用编程。如果涉及的软件采用面向对象编程，那么远程过程调用亦可称作远程调用或远程方法调用

目标

在工作队列模式中介绍了如何使用工作队列（work queue）在多个工作者（woker）中间分发耗时的任务。如果我们需要将一个函数运行在远程计算机上并且等待从那儿获取结果，这种模式通常被称为远程过程调用（Remote Procedure Call）或者RPC。

为了更好地描述这个问题，我们会使用RabbitMQ来构建一个RPC系统：包含一个客户端和一个RPC服务器。我们从远程服务器执行一条命令，返回给客户端

声明回调队列

一般来说通过RabbitMQ来实现RPC是很容易的。一个客户端发送请求信息，服务器端将其应用到一个回复信息中。为了接收到回复信息，客户端需要在发送请求的时候同时发送一个回调队列（callback queue）的地址（思考一下为什么要这么做？）。实现如下

result = channel.queue_declare(exclusive=True)
callback_queue = result.method.queue

channel.basic_publish(exchange='',
                      routing_key='rpc_queue',
                      properties=pika.BasicProperties(
                            reply_to = callback_queue,
                            ),
                      body=request)

消息属性

AMQP协议给消息预定义了一系列的14个属性。大多数属性很少会用到，除了以下几个：

delivery_mode（投递模式）：将消息标记为持久的（值为2）或暂存的（除了2之外的其他任何值）。工作队列模式中有介绍。
content_type（内容类型）:用来描述编码的mime-type。例如在实际使用中常常使用application/json来描述JOSN编码类型。
reply_to（回复目标）：通常用来命名回调队列。
correlation_id（关联标识）：用来将RPC的响应和请求关联起来。

RPC的工作步骤

1. 当客户端启动的时候，它创建一个匿名独享的回调队列。
2. 在RPC请求中，客户端发送带有两个属性的消息：一个是设置回调队列的 reply_to 属性，另一个是设置唯一值的 correlation_id 属性。
3. 将请求发送到一个 rpc_queue 队列中。
4. RPC工作者（又名：服务器）等待请求发送到这个队列中来。当请求出现的时候，它执行他的工作并且将带有执行结果的消息发送给reply_to字段指定的队列。
5. 客户端等待回调队列里的数据。当有消息出现的时候，它会检查correlation_id属性。如果此属性的值与请求匹配，将它返回给应用。
   correlation_id
   在实际使用中，客户端肯定会有多个RPC请求，为了区分这样请求。我们可能会想到给每个RPC请求新建一个回调队列（这样请求与回复一一对应），但是这样的话很占资源。

correlation_id的引入使这一情况得到解决。他允许每个客户端只建立一个独立的回调队列，然后给每个请求设置一个独一无二的值。稍后，当我们从回调队列中接收到一个消息的时候，我们就可以查看这条属性从而将响应和请求匹配起来。如果我们接手到的消息的correlation_id是未知的，那就直接销毁掉它，因为它不属于我们的任何一条请求。

代码实现

服务端

# 1. 声明接收指令的队列名rpc_queue
# 2. 开始监听队列，收到消息后 调用on_request函数
# 3. 把run_cmd执行结果，发送回客户端指定的reply_to 队列
import subprocess
import pika
import time

credentials = pika.PlainCredentials('zou', '123')

parameters = pika.ConnectionParameters(host='192.168.56.10', credentials=credentials)
connection = pika.BlockingConnection(parameters)

channel = connection.channel()  # 队列连接通道

channel.queue_declare(queue='rpc_queue2')


def run_cmd(cmd):
    cmd_obj = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    result = cmd_obj.stdout.read() + cmd_obj.stderr.read()

    return result


def on_request(ch, method, props, body):
    cmd = body.decode("utf-8")

    print(" [.] run (%s)" % cmd)
    response = run_cmd(cmd)

    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_consume(on_request, queue='rpc_queue2')

print(" [x] Awaiting RPC requests")
channel.start_consuming()

客户端

# 1.声明一个队列，作为reply_to返回消息结果的队列
# 2.  发消息到队列，消息里带一个唯一标识符uid，reply_to
# 3.  监听reply_to 的队列，直到有结果
import queue

import pika
import uuid

class CMDRpcClient(object):
    def __init__(self):
        credentials = pika.PlainCredentials('zou', '123')
        parameters = pika.ConnectionParameters(host='192.168.56.10',credentials=credentials)
        self.connection = pika.BlockingConnection(parameters)
        self.channel = self.connection.channel()

        result = self.channel.queue_declare(exclusive=True)
        self.callback_queue = result.method.queue #命令的执行结果的queue

        # 声明要监听callback_queue,no_ack=True，标识码不符合立即删除该消息。
        self.channel.basic_consume(self.on_response, no_ack=True,queue=self.callback_queue)



    def on_response(self, ch, method, props, body):
        """
        收到服务器端命令结果后执行这个函数

        """
        if self.corr_id == props.correlation_id:
            self.response = body.decode("utf-8") #把执行结果赋值给Response

    def call(self, n):
        self.response = None
        self.corr_id = str(uuid.uuid4()) #唯一标识符号
        self.channel.basic_publish(exchange='',
                                   routing_key='rpc_queue2',
                                   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()  # 检测监听的队列里有没有新消息，如果有，收，如果没有，返回None
            #检测有没有要发送的新指令
        return self.response

cmd_rpc = CMDRpcClient()

print(" [x] Requesting fib(30)")
response = cmd_rpc.call('pwd')

print(response)

运行结果

客户端代码讲解

1. 建立连接、通道并且为回复（reply）声明独享的回调队列。
2. 我们订阅这个回调队列，以便接收RPC的响应。
3. “on_response”回调函数对每一个响应执行一个非常简单的操作，检查每一个响应消息的correlation_id属性是否与我们期待的一致，如果一致，将响应结果赋给self.response，然后跳出consuming循环。
4. 接下来，我们定义我们的主要方法 call 方法。它执行真正的RPC请求。
   在这个方法中，首先我们生成一个唯一的 correlation_id 值并且保存起来，'on_response'回调函数会用它来获取符合要求的响应。
5. 接下来，我们将带有 reply_to 和 correlation_id 属性的消息发布出去。
   最后，将响应返回给用户。
6. 我们可以看到客户端和服务端都既是订阅者，也是发布者。那么对于客户端的rpc请求，如果 RPC服务器运行的过慢。那么我们可以类似“工作队列”模式，运行多个server进行请求的处理。