基于python實(shí)現(xiàn)RPC的demo

這是一個(gè)遠(yuǎn)程過程調(diào)用（RPC）的實(shí)現(xiàn)demo，可以實(shí)現(xiàn)不同的python進(jìn)程之間通信和互相調(diào)用函數(shù)，簡單易用，易于擴(kuò)展。更多功能也可進(jìn)一步完善，本文介紹了該實(shí)現(xiàn)的主要思路。

前言

計(jì)劃手?jǐn)]一個(gè)rpc甚久了，在間歇性push自己下終于完成的差不多了。寫這個(gè)demo的原因，1）是為了學(xué)習(xí)與思考下這部分主體功能和實(shí)現(xiàn)思路，2）是調(diào)包時(shí)可以毫無心理負(fù)擔(dān)，并產(chǎn)生一種不過如此的優(yōu)越感。
實(shí)現(xiàn)這部分內(nèi)容主要依據(jù)的還是自己的想法，因此可能會(huì)有bug或者有更好的實(shí)現(xiàn)方式，僅供學(xué)習(xí)和參考，完整代碼可參考Gitee鏈接。
實(shí)現(xiàn)的時(shí)候用的是python2.7，忘記換了，下次一定更新。

一、主要內(nèi)容

所謂RPC，是遠(yuǎn)程過程調(diào)用（Remote Procedure Call）的簡寫，網(wǎng)上解釋很多，簡單來說，就是在當(dāng)前進(jìn)程調(diào)用其他進(jìn)程的函數(shù)時(shí)，體驗(yàn)就像是調(diào)用本地寫的函數(shù)一般。
本文實(shí)現(xiàn)的是在本地調(diào)用遠(yuǎn)端的類class對(duì)象的接口，也就是本地的client不實(shí)例化類對(duì)象，調(diào)用的是server端的類對(duì)象接口。
為了達(dá)到讓調(diào)用層無須關(guān)心底層實(shí)現(xiàn)，擁有絲滑般的體驗(yàn)，就需要以下幾個(gè)部分：

• 客戶端需要把類的接口提取出來，并將調(diào)用函數(shù)事件捕獲存儲(chǔ)起來；服務(wù)端需要把類的公有函數(shù)作為可遠(yuǎn)程調(diào)用的接口。
• 客戶端把調(diào)用函數(shù)的事件（調(diào)用的函數(shù)，參數(shù)）進(jìn)行序列化并發(fā)送給服務(wù)端；服務(wù)端將客戶端的調(diào)用事件反序列化，并執(zhí)行相應(yīng)的接口，將返回值發(fā)送給客戶端。
• 客戶端與服務(wù)端通過某種方式（一般就是網(wǎng)絡(luò)socket）進(jìn)行通信。

在下面時(shí)序圖的灰色部分，對(duì)于調(diào)用方來說是透明的，它的執(zhí)行結(jié)果應(yīng)該和執(zhí)行本地的函數(shù)時(shí)一致的。

二、實(shí)現(xiàn)步驟

1. 進(jìn)程間的通信

本文采用了基于TCP的sokcet連接來進(jìn)行進(jìn)程之間的通信，更多實(shí)現(xiàn)細(xì)節(jié)可參考之前博客。
在此需要注意：

本文采用了select模塊來監(jiān)聽網(wǎng)絡(luò)事件，如果服務(wù)端未收到任何的網(wǎng)絡(luò)消息會(huì)一直阻塞在這兒。如果服務(wù)端除了提供rpc調(diào)用服務(wù)之外還需要執(zhí)行其他邏輯，那么應(yīng)當(dāng)采用非阻塞，輪詢socket的方式來判斷是否有新的網(wǎng)絡(luò)事件。

# ServerBase.py
def process(self):
    readable, writable, exceptional = select.select(self.inputs, self.outputs, self.conns.values())
    for conn in readable:
        if conn is self.socket:
            self._handle_conn()
        else:
            self._handle_recv(conn)
    for conn in writable:
        pass
    for conn in exceptional:
        self._handle_leave(conn)

客戶端的網(wǎng)絡(luò)事件本文通過創(chuàng)建新的線程來監(jiān)聽的。并不會(huì)影響客戶端主線程的執(zhí)行，因此可以盡情的阻塞。部分代碼如下：

# AsynCallback.py
class AsyncTaskManager(object):
? ? _asy_events = dict()

? ? def __init__(self, loop, *args):
? ? ? ? super(AsyncTaskManager, self).__init__()
? ? ? ? self._loop_fun = loop

? ? def __call__(self, *args, **kwargs):
? ? ? ? proc = threading.Thread(target=self._exec_loop, args=args, kwargs=kwargs)
? ? ? ? proc.start()

? ? def _exec_loop(self, *args, **kwargs):
? ? ? ? while True:
? ? ? ? ? ? net_resp = self._loop_fun(*args, **kwargs)
? ? ? ? ? ? for resp in net_resp:
? ? ? ? ? ? ? ? asy_event = self._asy_events.pop(resp.rid)
? ? ? ? ? ? ? ? asy_event.set()

# Client.py
class Client(TaskHandle, ClientBase):

?? ?@AsyncTaskManager
?? ?def process(self):
?? ? ? ?super(Client, self).process()
?? ? ? ?_events = []
?? ? ? ?while self.has_events:
?? ? ? ? ? ?event = self.get_next_event()
?? ? ? ? ? ?data = event[1]
?? ? ? ? ? ?_events.append(self.unpack_respond(data))
?? ? ? ?return _events

序列化方式，本文采用了庫pickle進(jìn)行序列化與反序列化，使用它的原因是可以將自定義類對(duì)象也進(jìn)行序列化，非常之高級(jí)。

2. 異步回調(diào)實(shí)現(xiàn)思路

對(duì)于需要返回值的函數(shù)調(diào)用，處理起來比較簡單，只需要將主線程阻塞等待，直至超時(shí)或者接收到了對(duì)應(yīng)函數(shù)的返回值即可。本文采用了threading.Event來阻塞與喚醒調(diào)用的函數(shù)，同時(shí)采用了裝飾器來實(shí)現(xiàn)這功能。若日后有更好的方法，可以輕易進(jìn)行替換。相關(guān)示例代碼如下所示：

@AsyncTaskManager.respond
def _handle_response(self, tid):
? ? """ 處理有返回值的情況
? ? 會(huì)阻塞線程直至收到返回值
? ? """
? ? task = self.pop_task(tid)
? ? if task.callback:
? ? ? ? task.callback()
? ? return self.pop_respond(tid)

@staticmethod
def respond(func):
? ? @wraps(func)
? ? def make_resp(handle, tid):
? ? ? ? """ 需要注意的是，和裝飾的函數(shù)參數(shù)含義需一致 """
? ? ? ? event = threading.Event()
? ? ? ? AsyncTaskManager._asy_events[tid] = event
? ? ? ? event.wait(timeout=TIME_OUT)
? ? ? ? return func(handle, tid)?? ?# 這兒才是真正執(zhí)行_handle_response的地方
? ? return make_resp

在實(shí)際的應(yīng)用過程中，應(yīng)有這樣的情況，服務(wù)端與客戶端都是獨(dú)立的應(yīng)用，通過rpc函數(shù)進(jìn)行通信和交互，而并不是某方為另外一方提供服務(wù)，那么此時(shí)返回值并不必要，只需要將要做的事通知另一方即可。對(duì)于此種情況，可以采用異步回調(diào)的方式來告知調(diào)用方對(duì)應(yīng)函數(shù)執(zhí)行成功了。

在文中依舊采用線程來完成該功能，客戶端調(diào)用函數(shù)之后創(chuàng)建一個(gè)新線程并阻塞住，等待服務(wù)端將執(zhí)行結(jié)果發(fā)回后再喚醒，如果有回調(diào)函數(shù)就執(zhí)行。示例代碼如下：

@AsyncTaskManager.callback
def _handle_call_back(self, tid):
? ? """ 處理有回調(diào)函數(shù)的調(diào)用
? ? callback會(huì)等tid事件調(diào)用成功之后 才會(huì)回調(diào)，且不會(huì)有返回值
? ? """
? ? task = self.pop_task(tid)
? ? if task.callback:
? ? ? ? task.callback()
? ? ? ??
@staticmethod
def callback(func):
? ? @wraps(func)
? ? def make_thread(event, *args, **kwargs):
? ? ? ? event.wait(timeout=TIME_OUT)
? ? ? ? func(*args, **kwargs)

? ? def make_async(handle, tid):
? ? ? ? """ 注意點(diǎn)同上 """
? ? ? ? event = threading.Event()
? ? ? ? AsyncTaskManager._asy_events[tid] = event
? ? ? ? _task = threading.Thread(target=lambda: make_thread(event, handle, tid))

? ? return make_async

總結(jié)

到此這篇關(guān)于基于python實(shí)現(xiàn)rpc遠(yuǎn)程過程調(diào)用的文章就介紹到這了,更多相關(guān)python rpc遠(yuǎn)程調(diào)用內(nèi)容請(qǐng)搜索腳本之家以前的文章或繼續(xù)瀏覽下面的相關(guān)文章希望大家以后多多支持腳本之家！

最新評(píng)論