Python中如何实现基于队列的远程调用

pip install pika

import pika

# 定义一个RPC处理类，用于接收远程调用请求，并返回结果
class RpcServer:
    def __init__(self):
        # 建立到RabbitMQ的连接
        self.connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
        self.channel = self.connection.channel()

        # 声明远程调用的队列
        self.channel.queue_declare(queue='rpc_queue')

        # 设置回调函数，当有请求过来时会调用此函数进行处理
        self.channel.basic_qos(prefetch_count=1)
        self.channel.basic_consume(queue='rpc_queue', on_message_callback=self.on_request)

    # 处理请求并返回结果
    def on_request(self, ch, method, props, body):
        n = int(body)

        print(" [.] fibonacci(%s)" % n)
        response = self.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)

    def fib(self, n):
        if n == 0:
            return 0
        elif n == 1:
            return 1
        else:
            return self.fib(n-1) + self.fib(n-2)

    # 开始监听请求
    def start(self):
        print(" [x] Awaiting RPC requests")
        self.channel.start_consuming()

if __name__ == "__main__":
    server = RpcServer()
    server.start()

import pika
import uuid

# 定义一个RPC调用类，用于发送远程调用请求，并等待结果返回
class RpcClient:
    def __init__(self):
        # 建立到RabbitMQ的连接
        self.connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
        self.channel = self.connection.channel()

        # 声明一个独占队列，用于接收结果
        result = self.channel.queue_declare(queue='', exclusive=True)
        self.callback_queue = result.method.queue

        # 设置回调函数，当结果返回时会调用此函数
        self.channel.basic_consume(queue=self.callback_queue, on_message_callback=self.on_response)

    # 发送远程调用请求
    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)

    # 处理结果返回
    def on_response(self, ch, method, props, body):
        if self.corr_id == props.correlation_id:
            self.response = body

    # 关闭连接
    def close(self):
        self.connection.close()

if __name__ == "__main__":
    client = RpcClient()

    # 发送远程调用请求
    print(" [x] Requesting fib(30)")
    response = client.call("30")
    print(" [.] Got %r" % response)

    client.close()