【一】RPC介绍

【1】介绍

• RPC（Remote Procedure Call）是一种远程过程调用的协议，它允许一个计算机程序通过网络请求调用远程服务器上的一个子程序或函数。
• 基于RabbitMQ实现的RPC可以更加可靠地实现远程过程调用。

【2】分布式的系统中使用

• 微服务之间的调用
• resful的接口
• rpc调用

【二】基于RabbitMQ实现RPC

【1】服务端

import pika

# 【1】连接 RabbitMQ 服务器
credentials = pika.PlainCredentials("admin","admin")
connection = pika.BlockingConnection(pika.ConnectionParameters('101.133.225.166',credentials=credentials))
channel = connection.channel()

# 【2】监听任务队列
channel.queue_declare(queue='rpc_queue')

# 【3】处理请求的回调函数
def on_request(ch, method, props, body):
    # 【3.1】将接收到的请求消息转换为整数
    n = int(body)
    
    # 【3.2】进行计算，这里示例是将输入的数字加上100
    response = n + 100
    
    # 【3.3】发送结果到 props.reply_to 所指定的队列中
    # 【3.4】correlation_id 是客户端传过来的任务ID
    # props.reply_to  要放结果的队列.
    # props.correlation_id  任务
    ch.basic_publish(exchange='',
                     routing_key=props.reply_to,
                     properties=pika.BasicProperties(correlation_id= props.correlation_id),
                     body=str(response))
    # 【4】确认消息已被处理
    ch.basic_ack(delivery_tag=method.delivery_tag)

# 【5】闲置消费
channel.basic_qos(prefetch_count=1)
# 【6】注册处理请求的回调函数
channel.basic_consume( queue='rpc_queue',on_message_callback=on_request,)
# 【7】开始消费消息
channel.start_consuming()

【2】客户端

import pika
import uuid

class FibonacciRpcClient(object):
    def __init__(self):
        # 连接 RabbitMQ 服务器
        credentials = pika.PlainCredentials("admin", "admin")
        self.connection = pika.BlockingConnection(pika.ConnectionParameters('101.133.225.166', credentials=credentials))
        self.channel = self.connection.channel()

        # 随机生成一个消息队列(用于接收结果)
        result = self.channel.queue_declare(queue='',exclusive=True)
        self.callback_queue = result.method.queue

        # 监听消息队列中是否有值返回,如果有值则执行 on_response 函数(一旦有结果,则执行on_response)
        self.channel.basic_consume(queue=self.callback_queue,on_message_callback=self.on_response, auto_ack=True)

    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())

        # 客户端 给 服务端 发送一个任务:  任务id = corr_id / 任务内容 = '30' / 用于接收结果的队列名称
        self.channel.basic_publish(exchange='',
                                   routing_key='rpc_queue', # 服务端接收任务的队列名称
                                   properties=pika.BasicProperties(
                                         reply_to = self.callback_queue, # 用于接收结果的队列
                                         correlation_id = self.corr_id, # 任务ID
                                         ),
                                   body=str(n))
        while self.response is None:
            self.connection.process_data_events()

        return self.response

fibonacci_rpc = FibonacciRpcClient()

response = fibonacci_rpc.call(50)
print('返回结果:',response)

• 服务端等待客户端发送任务请求，接收到任务后执行相应的逻辑计算，并将结果通过props.reply_to指定的队列发送回去。
• 客户端在调用远程函数时，会生成一个唯一的任务ID，并将任务信息发送给服务端。
  • 然后通过一个专门用于接收结果的队列等待服务端返回结果。
• 这种基于RabbitMQ实现的RPC可以更好地解决分布式系统中不同服务间的通信问题。
  • 例如，假设存在一个订单系统和一个支付系统，订单系统需要调用支付系统进行支付操作。
  • 通过RPC，可以在订单系统中直接发起支付请求，并通过RPC的方式将支付结果返回给订单系统，实现了不同系统之间的解耦和高效通信。