RPCHandler 和 RPCProxy 的基本思路是很比較簡單的。 如果一個客戶端想要調(diào)用一個遠程函數(shù)，比如 foo(1, 2, z=3) ,代理類創(chuàng)建一個包含了函數(shù)名和參數(shù)的元組 (‘foo’, (1, 2), {‘z’: 3}) 。 這個元組被pickle序列化后通過網(wǎng)絡(luò)連接發(fā)生出去。 這一步在 RPCProxy 的 getattr() 方法返回的 do_rpc() 閉包中完成。

服務(wù)器接收后通過pickle反序列化消息，查找函數(shù)名看看是否已經(jīng)注冊過，然后執(zhí)行相應(yīng)的函數(shù)。 執(zhí)行結(jié)果(或異常)被pickle序列化后返回發(fā)送給客戶端。實例需要依賴 multiprocessing 進行通信。 不過，這種方式可以適用于其他任何消息系統(tǒng)。例如，如果你想在ZeroMQ之上實習(xí)RPC， 僅僅只需要將連接對象換成合適的ZeroMQ的socket對象即可。

先實現(xiàn)server端

import jsonfrom multiprocessing.connection import Listenerfrom threading import Threadclass RPCHandler: def __init__(self): self._functions = {} def register_function(self, func): self._functions[func.__name__] = func def handle_connection(self, connection): try: while True:func_name, args, kwargs = json.loads(connection.recv())# Run the RPC and send a responsetry: r = self._functions[func_name](*args, **kwargs) connection.send(json.dumps(r))except Exception as e: connection.send(json.dumps(e)) except EOFError: passdef rpc_server(handler, address, authkey): sock = Listener(address, authkey=authkey) while True: client = sock.accept() t = Thread(target=handler.handle_connection, args=(client,)) t.daemon = True t.start()# Some remote functionsdef add(x,y): return x+yif __name__ == ’__main__’: handler = RPCHandler() handler.register_function(add) # Run the server rpc_server(handler, (’127.0.0.1’, 17000), authkey=b’peekaboo’)

再實現(xiàn)client端

import jsonfrom multiprocessing.connection import Clientclass RPCProxy: def __init__(self, connection): self._connection = connection def __getattr__(self, name): def do_rpc(*args, **kwargs): self._connection.send(json.dumps((name, args, kwargs))) result = json.loads(self._connection.recv()) if isinstance(result, Exception):raise result return result return do_rpcif __name__ == ’__main__’: c = Client((’127.0.0.1’, 17000), authkey=b’peekaboo’) proxy = RPCProxy(c) res = proxy.add(2, 3) print(res)

以上就是本文的全部內(nèi)容，希望對大家的學(xué)習(xí)有所幫助，也希望大家多多支持好吧啦網(wǎng)。