快捷導(dǎo)航

Python如何實(shí)現(xiàn)遠(yuǎn)程方法調(diào)用

更新時(shí)間：2020年08月07日 09:16:54 作者：David Beazley

這篇文章主要介紹了Python如何實(shí)現(xiàn)遠(yuǎn)程方法調(diào)用，文中講解非常細(xì)致，幫助大家更好的理解和學(xué)習(xí)，感興趣的朋友可以了解下

問(wèn)題

你想在一個(gè)消息傳輸層如 sockets 、multiprocessing connections 或 ZeroMQ 的基礎(chǔ)之上實(shí)現(xiàn)一個(gè)簡(jiǎn)單的遠(yuǎn)程過(guò)程調(diào)用（RPC）。

解決方案

將函數(shù)請(qǐng)求、參數(shù)和返回值使用pickle編碼后，在不同的解釋器直接傳送pickle字節(jié)字符串，可以很容易的實(shí)現(xiàn)RPC。下面是一個(gè)簡(jiǎn)單的PRC處理器，可以被整合到一個(gè)服務(wù)器中去：

# rpcserver.py

import pickle
class RPCHandler:
  def __init__(self):
    self._functions = { }

  def register_function(self, func):
    self._functions[func.__name__] = func

  def handle_connection(self, connection):
    try:
      while True:
        # Receive a message
        func_name, args, kwargs = pickle.loads(connection.recv())
        # Run the RPC and send a response
        try:
          r = self._functions[func_name](*args,**kwargs)
          connection.send(pickle.dumps(r))
        except Exception as e:
          connection.send(pickle.dumps(e))
    except EOFError:
       pass

要使用這個(gè)處理器，你需要將它加入到一個(gè)消息服務(wù)器中。你有很多種選擇，但是使用 multiprocessing 庫(kù)是最簡(jiǎn)單的。下面是一個(gè)RPC服務(wù)器例子：

from multiprocessing.connection import Listener
from threading import Thread

def 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 functions
def add(x, y):
  return x + y

def sub(x, y):
  return x - y

# Register with a handler
handler = RPCHandler()
handler.register_function(add)
handler.register_function(sub)

# Run the server
rpc_server(handler, ('localhost', 17000), authkey=b'peekaboo')

為了從一個(gè)遠(yuǎn)程客戶(hù)端訪問(wèn)服務(wù)器，你需要?jiǎng)?chuàng)建一個(gè)對(duì)應(yīng)的用來(lái)傳送請(qǐng)求的RPC代理類(lèi)。例如

import pickle

class RPCProxy:
  def __init__(self, connection):
    self._connection = connection
  def __getattr__(self, name):
    def do_rpc(*args, **kwargs):
      self._connection.send(pickle.dumps((name, args, kwargs)))
      result = pickle.loads(self._connection.recv())
      if isinstance(result, Exception):
        raise result
      return result
    return do_rpc

要使用這個(gè)代理類(lèi)，你需要將其包裝到一個(gè)服務(wù)器的連接上面，例如：

>>> from multiprocessing.connection import Client
>>> c = Client(('localhost', 17000), authkey=b'peekaboo')
>>> proxy = RPCProxy(c)
>>> proxy.add(2, 3)

5
>>> proxy.sub(2, 3)
-1
>>> proxy.sub([1, 2], 4)
Traceback (most recent call last):
 File "<stdin>", line 1, in <module>
 File "rpcserver.py", line 37, in do_rpc
  raise result
TypeError: unsupported operand type(s) for -: 'list' and 'int'
>>>

要注意的是很多消息層（比如 multiprocessing ）已經(jīng)使用pickle序列化了數(shù)據(jù)。如果是這樣的話(huà)，對(duì) pickle.dumps() 和 pickle.loads() 的調(diào)用要去掉。

討論

RPCHandler 和 RPCProxy 的基本思路是很比較簡(jiǎn)單的。如果一個(gè)客戶(hù)端想要調(diào)用一個(gè)遠(yuǎn)程函數(shù)，比如 foo(1, 2, z=3) ,代理類(lèi)創(chuàng)建一個(gè)包含了函數(shù)名和參數(shù)的元組 ('foo', (1, 2), {'z': 3}) 。這個(gè)元組被pickle序列化后通過(guò)網(wǎng)絡(luò)連接發(fā)生出去。這一步在 RPCProxy 的 __getattr__() 方法返回的 do_rpc() 閉包中完成。服務(wù)器接收后通過(guò)pickle反序列化消息，查找函數(shù)名看看是否已經(jīng)注冊(cè)過(guò)，然后執(zhí)行相應(yīng)的函數(shù)。執(zhí)行結(jié)果(或異常)被pickle序列化后返回發(fā)送給客戶(hù)端。我們的實(shí)例需要依賴(lài) multiprocessing 進(jìn)行通信。不過(guò)，這種方式可以適用于其他任何消息系統(tǒng)。例如，如果你想在ZeroMQ之上實(shí)習(xí)RPC，僅僅只需要將連接對(duì)象換成合適的ZeroMQ的socket對(duì)象即可。

由于底層需要依賴(lài)pickle，那么安全問(wèn)題就需要考慮了（因?yàn)橐粋€(gè)聰明的黑客可以創(chuàng)建特定的消息，能夠讓任意函數(shù)通過(guò)pickle反序列化后被執(zhí)行）。因此你永遠(yuǎn)不要允許來(lái)自不信任或未認(rèn)證的客戶(hù)端的RPC。特別是你絕對(duì)不要允許來(lái)自Internet的任意機(jī)器的訪問(wèn)，這種只能在內(nèi)部被使用，位于防火墻后面并且不要對(duì)外暴露。

作為pickle的替代，你也許可以考慮使用JSON、XML或一些其他的編碼格式來(lái)序列化消息。例如，本機(jī)實(shí)例可以很容易的改寫(xiě)成JSON編碼方案。還需要將 pickle.loads() 和 pickle.dumps() 替換成 json.loads() 和 json.dumps() 即可：

# jsonrpcserver.py
import json

class RPCHandler:
  def __init__(self):
    self._functions = { }

  def register_function(self, func):
    self._functions[func.__name__] = func

  def handle_connection(self, connection):
    try:
      while True:
        # Receive a message
        func_name, args, kwargs = json.loads(connection.recv())
        # Run the RPC and send a response
        try:
          r = self._functions[func_name](*args,**kwargs)
          connection.send(json.dumps(r))
        except Exception as e:
          connection.send(json.dumps(str(e)))
    except EOFError:
       pass

# jsonrpcclient.py
import json

class 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())
      return result
    return do_rpc

實(shí)現(xiàn)RPC的一個(gè)比較復(fù)雜的問(wèn)題是如何去處理異常。至少，當(dāng)方法產(chǎn)生異常時(shí)服務(wù)器不應(yīng)該奔潰。因此，返回給客戶(hù)端的異常所代表的含義就要好好設(shè)計(jì)了。如果你使用pickle，異常對(duì)象實(shí)例在客戶(hù)端能被反序列化并拋出。如果你使用其他的協(xié)議，那得想想另外的方法了。不過(guò)至少，你應(yīng)該在響應(yīng)中返回異常字符串。我們?cè)贘SON的例子中就是使用的這種方式。

以上就是Python如何實(shí)現(xiàn)遠(yuǎn)程方法調(diào)用的詳細(xì)內(nèi)容，更多關(guān)于Python遠(yuǎn)程方法調(diào)用的資料請(qǐng)關(guān)注腳本之家其它相關(guān)文章！

您可能感興趣的文章: