第Python用內(nèi)置模塊來構(gòu)建REST服務(wù)與RPC服務(wù)實戰(zhàn)print(name,value)

測試HTTP客戶端,考慮使用httpbin服務(wù)()。這個站點會接收發(fā)出的請求，然后以JSON的形式將相應(yīng)信息回傳回來。

importrequests

r=requests.get('/getname=Daven=37',

...headers={'User-agent':'goaway/1.0'})

resp=r.json()

resp['headers']

{'Accept':'*/*','Accept-Encoding':'gzip,deflate','Host':'','User-Agent':'goaway/1.0','X-Amzn-Trace-Id':'Root=1-62708c06-7c7d8cc4441479c65faea5b4'}

通過XML-RPC實現(xiàn)簡單的遠(yuǎn)程調(diào)用

RPC,通俗的講，想找到一個方式去運行在遠(yuǎn)程機(jī)器上面的Python程序中的函數(shù)或方法。

實現(xiàn)一個遠(yuǎn)程方法調(diào)用的最簡單方式是使用XML-RPC。下面實現(xiàn)了鍵值存儲功能的簡單RPC服務(wù)器：

"""

@File:app.py

@Author:LiRuilong

@Version:1.0

@Desc:None

#hereputtheimportlib

fromxmlrpc.serverimportSimpleXMLRPCServer

classKeyValueServer:

_rpc_methods_=['get','set','delete','exists','keys']

def__init__(self,address):

self._data={}

self._serv=SimpleXMLRPCServer(address,allow_none=True)

#注冊方法

fornameinself._rpc_methods_:

self._serv.register_function(getattr(self,name))

defget(self,name):

returnself._data[name]

defset(self,name,value):

self._data[name]=value

defdelete(self,name):

delself._data[name]

defexists(self,name):

returnnameinself._data

defkeys(self):

returnlist(self._data)

defserve_forever(self):

self._serv.serve_forever()

#Example

if__name__=='__main__':

kvserv=KeyValueServer(('',15001))

kvserv.serve_forever()

RPC客戶端測試

PSE:\dockerpython

Python3.9.0(tags/v3.9.0:9cf6752,Oct52025,15:23:07)[MSCv.192732bit(Intel)]onwin32

Type"help","copyright","credits"or"license"formoreinformation.

fromxmlrpc.clientimportServerProxy

s=ServerProxy('http://localhost:15001',allow_none=True)

s.set('foo','bar')

s.set('spam',[1,2,3])

s.keys()

['foo','spam']

s.get('foo')

'bar'

s.get('spam')

[1,2,3]

s.delete('spam')

s.exists('spam')

False

XML-RPC可以讓很容易的構(gòu)造一個簡單的遠(yuǎn)程調(diào)用服務(wù)。所需要做的僅僅是創(chuàng)建一個服務(wù)器實例，通過它的方法register_function()來注冊函數(shù)，然后使用方法serve_forever()啟動它。在上面將這些步驟放在一起寫到一個類中

這并不是必須的。還可以像下面這樣創(chuàng)建一個服務(wù)器：

fromxmlrpc.serverimportSimpleXMLRPCServer

fromxmlrpc.serverimportSimpleXMLRPCServer

defadd(x,y):

returnx+y

serv=SimpleXMLRPCServer(('',15000))

serv.register_function(add)

serv.serve_forever()

XML-RPC暴露出來的函數(shù)只能適用于部分?jǐn)?shù)據(jù)類型，比如字符串、整形、列表和字典,不應(yīng)該將XML-RPC服務(wù)以公共API的方式暴露出來。

XML-RPC的一個缺點是它的性能。SimpleXMLRPCServer的實現(xiàn)是單線程的，所以它不適合于大型程序

由于XML-RPC將所有數(shù)據(jù)都序列化為XML格式，所以它會比其他的方式運行的慢一些。但是它也有優(yōu)點，這種方式的編碼可以被絕大部分其他編程語言支持。通過使用這種方式，其他語言的客戶端程序都能訪問的服務(wù)。

通過multiprocessing實現(xiàn)RPC調(diào)用

在一個消息傳輸層如sockets、multiprocessing.connections或zeroMQ的基礎(chǔ)之上實現(xiàn)一個簡單的遠(yuǎn)程過程調(diào)用(RPC)

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

RPC服務(wù)端

"""

@File:rpcserver.py

@Author:LiRuilong

@Version:1.0

@Desc:遠(yuǎn)程調(diào)用服務(wù)

#hereputtheimportlib

importpickle

frommultiprocessing.connectionimportListener

fromthreadingimportThread

@Time:2025/07/0820:28:02

@Author:LiRuilong

@Version:1.0

@Desc:None

Args:

遠(yuǎn)程調(diào)用處理器

Returns:

void

classRPCHandler:

def__init__(self):

self._functions={}

@Time:2025/07/0820:16:47

@Author:LiRuilong

@Version:1.0

@Desc:函數(shù)注冊

Args:

func

Returns:

void

defregister_function(self,func):

self._functions[func.__name__]=func

@Time:2025/07/0820:17:51

@Author:LiRuilong

@Version:1.0

@Desc:調(diào)用函數(shù)

Args:

connection

Returns:

void

defhandle_connection(self,connection):

try:

whileTrue:

func_name,args,kwargs=pickle.loads(connection.recv())

try:

print("調(diào)用函數(shù)：",(func_name,args,kwargs))

r=self._functions[func_name](*args,**kwargs)

print("返回結(jié)果：",r)

connection.send(pickle.dumps(r))

exceptExceptionase:

connection.send(pickle.dumps(e))

exceptExceptionase:

pass

defrpc_server(handler,address,authkey):

sock=Listener(address,authkey=authkey)

whileTrue:

client=sock.accept()

t=Thread(target=handler.handle_connection,args=(client,))

t.daemon=True

print("函數(shù)開始執(zhí)行")

t.start()

defadd(x,y):

returnx+y

defsub(x,y):

returnx-y

if__name__=='__main__':

print(format("開始加載RPC處理器",'》20'))

handler=RPCHandler()

print(format("處理器加載完成，注冊函數(shù)",'》20'))

handler.register_function(add)

handler.register_function(sub)

print(format("函數(shù)注冊成功,服務(wù)啟動",'》20'))

rpc_server(handler,('localhost',17000),authkey=b'peekaboo')

RPC客戶端

importpickle

frommultiprocessing.connectionimportClient

classRPCProxy:

def__init__(self,connection):

self._connection=connection

def__getattr__(self,name):

print("開始調(diào)用函數(shù)",name)

defdo_rpc(*args,**kwargs):

self._connection.send(pickle.dumps((name,args,kwargs)))

result=pickle.loads(self._connection.recv())

print("返回結(jié)果",result)

ifisinstance(result,Exception):

raiseresult

returnresult

returndo_rpc

c=Client(('localhost',17000),authkey=b'peekaboo')

print(format("建立連接，創(chuàng)建RPC代理",'》30'),c)

proxy=RPCProxy(c)

print(format("創(chuàng)建代理成功",'》30'))

print("add(2,3)=",proxy.add(2,3))

print("sub(2,3)=",proxy.sub(2,3))

D:\python\Python310\python.exeD:/python/code/rabbit_mq_demo/rpcserver.py

開始加載RPC處理器》》》》》》》》》》

處理器加載完成，注冊函數(shù)》》》》》》》》

函數(shù)注冊成功,服務(wù)啟動》》》》》》》》》

函數(shù)開始執(zhí)行

調(diào)用函數(shù)：(add,(2,3),{})

返回結(jié)果：5

調(diào)用函數(shù)：(sub,(2,3),{})

返回結(jié)果：-1

==============

D:\python\Python310\python.exeD:/python/code/rabbit_mq_demo/RPC.py

建立連接，創(chuàng)建RPC代理》》》》》》》》》》》》》》》》》》multiprocessing.connection.Connectionobjectat0x00DFACA0

創(chuàng)建代理成功》》》》》》》》》》》》》》》》》》》》》》》》

開始調(diào)用函數(shù)add

返回結(jié)果5

add(2,3)=5

開始調(diào)用函數(shù)sub

返回結(jié)果-1

sub(2,3)=-1

Processfinishedwithexitcode0

RPCHandler和RPCProxy的基本思路是很比較簡單的。

如果一個客戶端想要調(diào)用一個遠(yuǎn)程函數(shù)，比如foo(1，2，z=3)，代理類創(chuàng)建一個包含了函數(shù)名和參數(shù)的元組(foo，(1，2)，{z：3})。這個元組被pickle序列化后通過網(wǎng)絡(luò)連接發(fā)生出去。

由于底層需要依賴pickle，那么安全問題就需要考慮了(因為一個聰明的黑客可以創(chuàng)建特定的消息，能夠讓任意函數(shù)通過pickle反序列化后被執(zhí)行)。

因此永遠(yuǎn)不要允許來自不信任或未認(rèn)證的客戶端的RPC。特別是絕對不要允許來自Internet的任意機(jī)器的訪問，這種只能在內(nèi)部被使用，位于防火墻后面并且不要對外暴露。

作為pickle的替代，也許可以考慮使用JSON、XML或一些其他的編碼格式來序列化消息。

例如，本機(jī)實例可以很容易的改寫成JSON編碼方案。還需要將pickle.1oads()和pickle.dumps()替換成json.1oads()和json.dumps()即可：

#hereputtheimportlib

importjson

........

defhandle_connection(self,connection):

try:

whileTrue:

#反序列化

func_name,args,kwargs=json.loads(connection.recv())

try:

print("調(diào)用函數(shù)：",(func_name,args,kwargs))

r=self._functions[func_name](*args,**kwargs)

print("返回結(jié)果：",r)

#序列化發(fā)送

connection.send(json.dumps(r))

exceptExceptionase:

connection.send(json.dumps(e))

exceptExceptionase:

pass

......

importjson

frommultiprocessing.connectionimportClient

classRPCProxy:

def__init__(self,connection):

self._connection=connection

def__getattr__(self,name):

print("開始調(diào)用函數(shù)",name)

defdo_rpc(*args,**kwargs):

print("JSON序列化后的值",json.dumps((name,args,kwargs)))

self._connection.send(json.dumps((name,args,kwargs)))

result=json.loads(self._connection.recv())

print("返回結(jié)果",result)

ifisinstance(result,Exception):

raiseresult

returnr