python系列之 RabbitMQ - RPC

迈不过友情╰ 2022-07-29 11:19 257阅读 0赞

远程过程调用（Remote procedure call (RPC))

在第二课我们学习了怎样使用 工作队列（work queues) 来在多个workers之间分发需要消时的 任务

但是如果我们需要在远程的服务器上调用一个函数并获取返回结果 我们需要怎么做呢？well这是一个不一样的故事。 这中模式通常被称为远程过程调用或RPC

在这一刻我们将要使用RabbitMQ来建立一个RPC系统：一个客户端和一个可扩展的RPC服务。由于我们没有任何耗时的任务值得分配，我们将要创建一个仿RPC服务并返回斐波纳契数值

## 客户端接口（Client interface） ##

为了阐明RPC服务怎么使用我们创建一个简单的客户端类，将用一个名为Call的方法发送一个RPC请求并阻塞直到获取结果：

fibonacci_rpc = FibonacciRpcClient()
    result = fibonacci_rpc.call(4)
    print("fib(4) is %r" % result)

## 回调队列（callback queue） ##

一般通过RabbitMQ执行RPC是很容易的。 一个客户端发送一个请求消息然后服务端返回一个消息作为应答。 为了接收返回消息客户端需要发送一个“callback" 队列请求地址，让我们试试：

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）或者 瞬时的（其它值）， 你需要记住这个属性（在第二课时用到过）

> content\_type : 用来描述 MIME 类型的编码 ,比如我们经常使用的 JSON 编码，设置这个属性就非常好实现： application/json

> reply\_to： 经常用来命名一个 callback 队列

> correlation\_id ： 用来关联RPC的请求与应答

## 关联ID （Correlation ID） ##

前面提到的方法我们建议为每个RPC请求创建一个callback队列。  那是相当低效的，但是幸好有一个更好的方法 -- 我们未每个客户端创建一个单独的callback队列。

但这带来了一个新的问题， 当在那个队列中接收了一个返回，我们并不清楚是这个结果时属于那个请求的，这样当correlation\_id属性使用后，我们为每个请求设置一个唯一值。然后当我们从callback队列中接收到一个消息后，我们查看一下这个属性，基于这个我们就能将请求和返回进行匹配。如果我们看到一个未知的correlation\_id值，我们可以安全的丢弃这个消息 -- 不属于我们的请求

你可能会问，为什么我们要在callback队列中忽略未知的消息，而不是通过这个错误执行失败？ 这是由于服务端的竞争条件的可能性（？？），虽然可能性不大,但在为请求发送ack消息之前,当发送给我们结果后RPC服务还是有死掉的可能。如果发生这样的情况，让重启RPC服务之后将会重新处理请求。 这就是为什么客户端必须妥善的处理重复响应。

## 概要(Summary) ##

![Center][]

我们的RPC将会这样执行：

>  当客户端启动后，它创建一个匿名的唯一的回调队列

> 对一个RPC请求, 客户端发送一个消息包含两个属性： reply\_to （用来设置回调队列）和 correlation\_id（用来为每个请求设置一个唯一标识）

> 请求发送到 rpc\_queue队列

> RPC worker( 服务端) 在那个队列中等待请求，当一个请求出现后，服务端就执行一个job并将结果消息发送给客户端，使用reply\_to字段中的队列

> 客户端在callback 队列中等待数据, 当一个消息出现后，检查这个correlation\_id属性,如果和请求中的值匹配将返回给应用

## 整合 ##

rpc\_server.py代码

#!/usr/bin/env python
    import pika
    
    connection = pika.BlockingConnection(pika.ConnectionParameters(
            host='localhost'))
    
    channel = connection.channel()
    
    channel.queue_declare(queue='rpc_queue')
    
    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)
        response = fib(n)
    
        ch.basic_publish(exchange='',
                         routing_key=props.reply_to,
                         properties=pika.BasicProperties(correlation_id = \
                                                             props.correlation_id),
                         body=str(response))
        ch.basic_ack(delivery_tag = method.delivery_tag)
    
    channel.basic_qos(prefetch_count=1)
    channel.basic_consume(on_request, queue='rpc_queue')
    
    print(" [x] Awaiting RPC requests")
    channel.start_consuming()

服务端代码详单简单：

> (4) 和往常一样我们建立一个连接并定义一个队列

> (11) 我们定义了   斐波纳契 函数，假定输入的都是合法正数

> (19) 我们定义了一个回调的 basic\_consume, RPC服务的核心。 当收到请求后执行这个函数并返回结果

> (32) 我们可能会执行多个服务端，为了在多个服务端上均匀的分布负荷，我们需要这是 prefetch\_count。

rpc\_client.py 代码：

#!/usr/bin/env python
    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):
            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)

客户端代码稍微复杂些：

> (7) 我们建立一个连接，通道并定义一个专门的’callback‘队列用来接收回复

> (16) 我们订阅了“callback”队列，因此我们能够接收 RPC 的返回结果

> (18) ’on\_response'  在每个返回中执行的回调是一个简单的job， 对每个返回消息将检查是否correlation\_id使我们需要查找的那个ID，如果是，将保存结果到 self.response 并终端consuming循环

> (23) 下一步，我们定义我们的main方法 - 执行实际的RPC请求

> (24) 在这方法中，首先我们生产一个唯一的 correlatin\_id 号并保存 -- 'on\_response"回调函数将用着号码来匹配发送和接收的消息值

> (25) 下一步，发布请求信息，使用两个属性： reply\_to 和 correlation\_id

> (32) 这一步我们可以坐等结果的返回

>(33) 最后我们返回结果给用户

我们的RPC服务现在已经就绪,可以开启服务：

$ python rpc_server.py
     [x] Awaiting RPC requests

请求一个斐波那契数，运行客户端