由于項(xiàng)目組現(xiàn)在用的rpc是基于google protobuf rpc協(xié)議實(shí)現(xiàn)的，所以花了點(diǎn)時(shí)間了解下protobuf rpc。rpc對(duì)于做分布式系統(tǒng)的人來(lái)說肯定不陌生，對(duì)于rpc不了解的童鞋可以自行g(shù)oogle，這里只是做個(gè)簡(jiǎn)單的介紹。rpc的主要功能是讓分布式系統(tǒng)的實(shí)現(xiàn)更為簡(jiǎn)單，為提供強(qiáng)大的遠(yuǎn)程調(diào)用而不損失本地調(diào)用語(yǔ)義的簡(jiǎn)潔性。為了實(shí)現(xiàn)這個(gè)目標(biāo)，rpc框架需要提供一種透明調(diào)用機(jī)制讓使用者不必顯示區(qū)分本地調(diào)用還是遠(yuǎn)程調(diào)用。rpc架構(gòu)涉及的組件如下：

客戶方像調(diào)用本地方法一樣去調(diào)用遠(yuǎn)程接口方法，RPC 框架提供接口的代理實(shí)現(xiàn)，實(shí)際的調(diào)用將委托給代理RpcProxy 。代理封裝調(diào)用信息并將調(diào)用轉(zhuǎn)交給RpcInvoker 去實(shí)際執(zhí)行。在客戶端的RpcInvoker 通過連接器RpcConnector 去維持與服務(wù)端的通道RpcChannel，并使用RpcProtocol 執(zhí)行協(xié)議編碼（encode）并將編碼后的請(qǐng)求消息通過通道發(fā)送給服務(wù)方。RPC 服務(wù)端接收器 RpcAcceptor 接收客戶端的調(diào)用請(qǐng)求，同樣使用RpcProtocol 執(zhí)行協(xié)議解碼（decode）。解碼后的調(diào)用信息傳遞給RpcProcessor 去控制處理調(diào)用過程，最后再委托調(diào)用給RpcInvoker 去實(shí)際執(zhí)行并返回調(diào)用結(jié)果。

protobuf rpc在上面組件中主要扮演RpcProtocol的角色，使得我們省去了協(xié)議的設(shè)計(jì)，并且protobuf協(xié)議在編碼和空間效率都是上非常高效的，這也是很多公司采用protobuf作為數(shù)據(jù)序列化和通信協(xié)議的原因。同時(shí)protobuf rpc定義了一個(gè)抽象的rpc框架，如下圖所示：

RpcServiceStub和RpcService類是protobuf編譯器根據(jù)proto定義生成的類，RpcService定義了服務(wù)端暴露給客戶端的函數(shù)接口，具體實(shí)現(xiàn)需要用戶自己繼承這個(gè)類來(lái)實(shí)現(xiàn)。RpcServiceStub定義了服務(wù)端暴露函數(shù)的描述，并將客戶端對(duì)RpcServiceStub中函數(shù)的調(diào)用統(tǒng)一轉(zhuǎn)換到調(diào)用RpcChannel中的CallMethod方法，CallMethod通過RpcServiceStub傳過來(lái)的函數(shù)描述符和函數(shù)參數(shù)對(duì)該次rpc調(diào)用進(jìn)行encode，最終通過RpcConnecor發(fā)送給服務(wù)方。對(duì)方以客戶端相反的過程最終調(diào)用RpcSerivice中定義的函數(shù)。事實(shí)上，protobuf rpc的框架只是RpcChannel中定義了空的CallMethod，所以具體怎樣進(jìn)行encode和調(diào)用RpcConnector都要自己實(shí)現(xiàn)。RpcConnector在protobuf中沒有定義，所以這個(gè)完成由用戶自己實(shí)現(xiàn)，它的作用就是收發(fā)rpc消息包。在服務(wù)端，RpcChannel通過調(diào)用RpcService中的CallMethod來(lái)具體調(diào)用RpcService中暴露給客戶端的函數(shù)。

介紹了這么多，對(duì)于怎么樣用protobuf rpc來(lái)實(shí)現(xiàn)一個(gè)rpc肯定還是一頭霧水吧，下面就用protobuf rpc來(lái)實(shí)現(xiàn)一個(gè)簡(jiǎn)單的python版rpc demo吧。

下面直接給出demo描述PRC的proto文件，至于proto文件的編寫規(guī)則可以參考protobuf官網(wǎng)。

common.proto文件：

package game;

message RequestMessage
{
  required string message = 1;
}

message ResponseMessage
{
  required string message = 1;
}

game_service.proto文件：

package game;

import "common.proto";
option py_generic_services = true;

service GameService
{
  rpc connect_server(RequestMessage) returns(RequestMessage);
}

common.proto文件描述了RPC中收發(fā)的消息；game_service.proto描述了服務(wù)器導(dǎo)出的connect_server函數(shù)，該函數(shù)接受RequestMessage對(duì)象作為參數(shù)，并返回RequestMessage對(duì)象。在使用PRC協(xié)議時(shí)，必須加上option py_generic_services  = true;可選項(xiàng)，要不然編譯器不會(huì)生成包含connect_server函數(shù)的GameService描述。

使用編譯器protoc編譯proto文件，具體命令為：
protoc.exe --python_out=. game_service.proto
編譯后生成的文件為game_service_pb2.py，該文件主要是實(shí)現(xiàn)了GameService和GameService_Stub類。GameService_Stub類用于客戶端調(diào)用者來(lái)調(diào)用GameService的服務(wù)。
前面已經(jīng)說了，在客戶端，RpcChannel只實(shí)現(xiàn)了一個(gè)空的CallMethod，所以需要繼承RpcChannel重新這個(gè)函數(shù)來(lái)encode消息和發(fā)送消息。在服務(wù)端RpcChannel需要調(diào)用CallMethod來(lái)調(diào)用Service中的函數(shù)。具體實(shí)現(xiàn)如下：

class MyRpcChannel(service.RpcChannel):
  def __init__(self, rpc_service, conn):
    super(MyRpcChannel, self).__init__()
    self.logger = LogManager.get_logger("MyRpcChannel")

  def CallMethod(self, method_descriptor, rpc_controller, request, response_class, done):
    """"protol buffer rpc 需要的函數(shù)，用來(lái)發(fā)送rpc調(diào)用"""
    self.logger.info('CallMethod')
    cmd_index = method_descriptor.index
    assert(cmd_index < 65535)
    data = request.SerializeToString()
    total_len = len(data) + 2
    self.conn.send_data(''.join([pack('<I', total_len), pack('<H', cmd_index), data]))

  def from_request(self):
    """"從網(wǎng)絡(luò)解析出一個(gè)完整的請(qǐng)求之后調(diào)的函數(shù)"""
    index_data = self.rpc_request.data[0:2]    
    cmd_index = unpack('<H', index_data)[0]  
    rpc_service = self.rpc_service
    s_descriptor = rpc_service.GetDescriptor()
    method = s_descriptor.methods[cmd_index]  
    try:
      request = rpc_service.GetRequestClass(method)()
      serialized = self.rpc_request.data[2:]    
      request.ParseFromString(serialized)  
      rpc_service.CallMethod(method, self.controller, request, None)
    except:
      self.logger.error("Call rpc method failed!")
      self.logger.log_last_except()
    return True

最后就是繼承GameService，并實(shí)現(xiàn)connect_server函數(shù)了。

class GameService(game_service_pb2.GameService):
  def __init__(self):
    self.logger = LogManager.get_logger("GameService")

  def connect_server(self, rpc_controller, request, callback):
    self.logger.info('%s', request.message)

 至于用于網(wǎng)絡(luò)收發(fā)消息的RpcConnector，可以使用python的asyncore庫(kù)實(shí)現(xiàn)，具體實(shí)現(xiàn)在這就不討論了。

從上面的實(shí)現(xiàn)來(lái)看，protobuf rpc的實(shí)現(xiàn)主要包括編寫proto文件并編譯生成對(duì)應(yīng)的service_pb2文件，繼承RpcChannel并實(shí)現(xiàn)CallMethod和調(diào)用Service的CallMethod，繼承Service來(lái)實(shí)現(xiàn)暴露給客戶端的函數(shù)。

以上就是本文的全部?jī)?nèi)容，希望對(duì)大家的學(xué)習(xí)有所幫助。

最新評(píng)論