利用Pika实现RPC

前面所叙述的消息大多都是单向传输的，也就是始终是从生产者传递向消费者。但是在RPC应用中，消息的传输是双向的，RPC客户端在调用RPC服务端的过程后，RPC服务端需要将执行结果反馈给RPC客户端，这个过程通常是异步的。

RPC的具体实现可参考前一节针对异步生产和消费的模式进行设计，这里仅给出用于实现RPC的关键代码。

首先是RPC服务器的设计。

class RPCServer:
	def __init__(self):
		# 这里省略Connection及Channel的创建
		self.consume = self.channel.basic_consume(self.on_request, queue='rpc_queue')
	
	def on_request(self, channel, method, props, body):
		# 接收并处理请求消息，形成返回消息
		channel.basic_publish(
            exchange='',
			routing_key=props.reply_to,
			properties=pika.BasicProperties(correlation_id=props.correlation_id),
			body=str(callback_msg)
        )
		channel.basic_ack(delivery_tag=method.delivery_tag)

然后是RPC客户端的设计。

class RPCClient:
	def __init__(self):
		# 这里省略Connection及Channel的创建
		result = self.channel.queue_declare(exclusive=True) # 不指定Queue的名字，消费者断开后自动删除Queue
		self.callback_queue = result.method.queue # 给Channel自动分配一个队列名
		
		# 注意这里不要设定自动确认消息，因为不能判定消息是否与请求匹配
		self.channel.basic_consume(
            self.on_response,
			no_ack=True,
			queue=self.callback_queue
        )
	
	def on_response(self, channel, method, props, body):
		if self.corr_id == props.correlation_id:
			# 如果RPCServer返回的结果的关联ID与请求ID相同，则缓存结果
			self.response = body.decode()
	
	def call(self, args):
		self.response = None
		# 生成一个请求ID
		self.corr_id = str(uuid.uuid1())
		
		# 使用默认Exchange向RPCServer中定义的rpc_queue发送信息，
		# 在properties中指定reply_to用于告知RPCServer响应消息的返回队列，
		# correlation_id用于匹配请求和响应
		self.channel.basic_publish(
            exchange='',
			routing_key='rpc_queue',
			properties=pika.BasicProperties(
				reply_to=self.callback_queue,
				correlation_id=self.corr_id
            ),
			body=args
        )
		
		while self.response is None:
			# 等待结果返回
			self.connection.process_data_events()

在设计RPC时，一般不会每一个RPC调用占用一个Queue，而是多个或者所有RPC调用共用一个Queue，消息通过BasicProperties中的correlation_id来区分RPC请求。