Python用内置模块来构建REST服务与RPC服务实战

第Python用内置模块来构建REST服务与RPC服务实战print(name,value)

测试HTTP客户端,考虑使用httpbin服务()。这个站点会接收发出的请求，然后以JSON的形式将相应信息回传回来。

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实现简单的远程调用

RPC,通俗的讲，想找到一个方式去运行在远程机器上面的Python程序中的函数或方法。

实现一个远程方法调用的最简单方式是使用XML-RPC。下面实现了键值存储功能的简单RPC服务器：

"""

@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可以让很容易的构造一个简单的远程调用服务。所需要做的仅仅是创建一个服务器实例，通过它的方法register_function()来注册函数，然后使用方法serve_forever()启动它。在上面将这些步骤放在一起写到一个类中

这并不是必须的。还可以像下面这样创建一个服务器：

fromxmlrpc.serverimportSimpleXMLRPCServer

fromxmlrpc.serverimportSimpleXMLRPCServer

defadd(x,y):

returnx+y

serv=SimpleXMLRPCServer(('',15000))

serv.register_function(add)

serv.serve_forever()

XML-RPC暴露出来的函数只能适用于部分数据类型，比如字符串、整形、列表和字典,不应该将XML-RPC服务以公共API的方式暴露出来。

XML-RPC的一个缺点是它的性能。SimpleXMLRPCServer的实现是单线程的，所以它不适合于大型程序

由于XML-RPC将所有数据都序列化为XML格式，所以它会比其他的方式运行的慢一些。但是它也有优点，这种方式的编码可以被绝大部分其他编程语言支持。通过使用这种方式，其他语言的客户端程序都能访问的服务。

通过multiprocessing实现RPC调用

在一个消息传输层如sockets、multiprocessing.connections或zeroMQ的基础之上实现一个简单的远程过程调用(RPC)

将函数请求、参数和返回值使用pickle编码后，在不同的解释器直接传送pickle字节字符串，可以很容易的实现RPC。下面是一个简单的PRC处理器，可以被整合到一个服务器中去：

RPC服务端

"""

@File:rpcserver.py

@Author:LiRuilong

@Version:1.0

@Desc:远程调用服务

#hereputtheimportlib

importpickle

frommultiprocessing.connectionimportListener

fromthreadingimportThread

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

@Author:LiRuilong

@Version:1.0

@Desc:None

Args:

远程调用处理器

Returns:

void

classRPCHandler:

def__init__(self):

self._functions={}

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

@Author:LiRuilong

@Version:1.0

@Desc:函数注册

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:调用函数

Args:

connection

Returns:

void

defhandle_connection(self,connection):

try:

whileTrue:

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

try:

print("调用函数：",(func_name,args,kwargs))

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

print("返回结果：",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("函数开始执行")

t.start()

defadd(x,y):

returnx+y

defsub(x,y):

returnx-y

if__name__=='__main__':

print(format("开始加载RPC处理器",'》20'))

handler=RPCHandler()

print(format("处理器加载完成，注册函数",'》20'))

handler.register_function(add)

handler.register_function(sub)

print(format("函数注册成功,服务启动",'》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("开始调用函数",name)

defdo_rpc(*args,**kwargs):

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

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

print("返回结果",result)

ifisinstance(result,Exception):

raiseresult

returnresult

returndo_rpc

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

print(format("建立连接，创建RPC代理",'》30'),c)

proxy=RPCProxy(c)

print(format("创建代理成功",'》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处理器》》》》》》》》》》

处理器加载完成，注册函数》》》》》》》》

函数注册成功,服务启动》》》》》》》》》

函数开始执行

调用函数：(add,(2,3),{})

返回结果：5

调用函数：(sub,(2,3),{})

返回结果：-1

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

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

建立连接，创建RPC代理》》》》》》》》》》》》》》》》》》multiprocessing.connection.Connectionobjectat0x00DFACA0

创建代理成功》》》》》》》》》》》》》》》》》》》》》》》》

开始调用函数add

返回结果5

add(2,3)=5

开始调用函数sub

返回结果-1

sub(2,3)=-1

Processfinishedwithexitcode0

RPCHandler和RPCProxy的基本思路是很比较简单的。

如果一个客户端想要调用一个远程函数，比如foo(1，2，z=3)，代理类创建一个包含了函数名和参数的元组(foo，(1，2)，{z：3})。这个元组被pickle序列化后通过网络连接发生出去。

由于底层需要依赖pickle，那么安全问题就需要考虑了(因为一个聪明的黑客可以创建特定的消息，能够让任意函数通过pickle反序列化后被执行)。

因此永远不要允许来自不信任或未认证的客户端的RPC。特别是绝对不要允许来自Internet的任意机器的访问，这种只能在内部被使用，位于防火墙后面并且不要对外暴露。

作为pickle的替代，也许可以考虑使用JSON、XML或一些其他的编码格式来序列化消息。

例如，本机实例可以很容易的改写成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("调用函数：",(func_name,args,kwargs))

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

print("返回结果：",r)

#序列化发送

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("开始调用函数",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("返回结果",result)

ifisinstance(result,Exception):

raiseresult

returnr