Python实现简易RPC框架：基于Socket与JSON的完整示例

陈默

发布时间：2026-10-14| 6人浏览过

原创

本文介绍如何使用Python Socket API和JSON序列化实现一个简易的RPC框架。通过继承机制封装TCPClient和RPCStub，客户端动态生成代理方法并序列化请求，服务端接收并反序列化执行对应函数。代码展示了从连接、注册方法到通信的完整流程，适合理解RPC基本原理。

Python实现简易RPC框架：基于Socket与JSON的完整示例

阅读本文前，建议具备基础的Python Socket网络编程知识。

回顾RPC基本概念

客户端（Client）：发起服务调用的主体。
客户端存根（Client Stub）：负责存储服务端地址，将客户端调用的参数打包为网络消息，并通过网络传输发送至服务端。
服务端存根（Server Stub）：接收客户端发送的消息，进行解包处理，随后调用本地实际服务逻辑进行处理。
服务端（Server）：真正提供服务逻辑的组件。
Network Service：底层传输层，支持TCP或HTTP协议。

实现JSON-RPC协议

整体设计思路

在编写代码前，先梳理整体设计思路。

1、Network Service部分直接利用Python Socket相关API实现；2、数据传输格式采用JSON，在Socket层会自动转换为二进制流，开发者无需手动处理二进制编码细节。

参考xmlrpc的设计模式，Client与Server均采用Mixin多继承机制构建，每个类职责单一，最终对外暴露的仅是一个封装后的类及其有限的方法接口。

Client端实现细节

首先从Client端代码开始实现。


# client.py

 

import rpcclient

 

c = rpcclient.RPCClient()

c.connect('127.0.0.1', 5000)

res = c.add(1, 2, c=3)

print(f'res: [{res}]')

使用示例为：实例化rpcclient.RPCClient类，调用connect方法连接Server端，随后直接调用Server端的add方法。该方法逻辑为将传入参数累加并返回结果，最后打印add方法的返回值。

RPCClient类同时继承自TCPClient类与RPCStub类。


# rpclient.py

class RPCClient(TCPClient, RPCStub):

    pass

其中TCPClient负责通过Socket建立TCP连接并发送数据请求，而RPCStub类主要负责将Client端调用Server端方法的相关信息打包，随后调用TCPClient类中的发送方法即可。这两个类的具体实现在rpclient.py文件中，代码如下。


class TCPClient(object):

    def __init__(self):

        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

 

    def connect(self, host, port):

        '''链接Server端'''

        self.sock.connect((host, port))

 

    def send(self, data):

        '''将数据发送到Server端'''

        self.sock.send(data)

 

    def recv(self, length):

        '''接受Server端回传的数据'''

        return self.sock.recv(length)

         

 

class RPCStub(object):

    def __getattr__(self, function):

        def _func(*args, **kwargs):

            d = {'method_name': function, 'method_args': args, 'method_kwargs': kwargs}

            self.send(json.dumps(d).encode('utf-8')) # 发送数据

            data = self.recv(1024) # 接收方法执行后返回的结果

            return data

 

        setattr(self, function, _func)

        return _func

TCPClient类仅涉及常规的Socket API操作，此处不再赘述，重点分析RPCStub类的实现。

当在Client端执行res = c.add(1, 2, c=3)时，实际触发RPCStub中的__getattr__方法。该方法提取Client端调用的方法名及参数，通过TCPServer类的send方法发送数据。发送前对数据进行JSON格式化，以便Server端解码，随后调用recv方法等待Server端返回的执行结果。

由于RPCClient类本身并未定义add方法，为实现用户直接在Client端调用Server端方法的体验，代码利用__getattr__动态构建_func方法，并通过setattr方法将其绑定到RPCClient类实例中。此时，该类便拥有了Server端方法的映射能力。

调用add方法时，实际执行对应的_func方法，将构造好的数据发送至Server端。

Server端实现与测试

Client端逻辑至此完成，接下来实现Server端，过程同样简洁。

Server端的使用方式如下。


# server.py

 

import rpcserver

 

def add(a, b, c=10):

    sum = a + b + c

    return sum

 

s = rpcserver.RPCServer()

s.register_function(add) # 注册方法

s.loop(5000) # 传入要监听的端口

实例化rpcserver.RPCServer类，通过register_function方法注册希望被Client端调用的函数，随后调用loop方法并传入监听端口。RPCServer类的具体实现如下。


# rpcserver.py

 

class RPCServer(TCPServer, JSONRPC, RPCStub):

    def __init__(self):

        TCPServer.__init__(self)

        JSONRPC.__init__(self)

        RPCStub.__init__(self)

 

    def loop(self, port):

        # 循环监听 5000 端口

        self.bind_listen(port)

        print('Server listen 5000 ...')

        while True:

            self.accept_receive_close()

 

    def on_msg(self, data):

        return self.call_method(data)

RPCServer类继承自TCPServer、JSONRPC及RPCStub，这些类的具体实现在rpcserver.py文件中，且代码已包含详细注释，故此处不再展开解释。


class TCPServer(object):

    def __init__(self):

        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)

 

    def bind_listen(self, port):

        self.sock.bind(('0.0.0.0', port))

        self.sock.listen(5)

 

    def accept_receive_close(self):

        '''获取Client端信息'''

        (client_socket, address) = self.sock.accept()

        msg = client_socket.recv(1024)

        data = self.on_msg(msg)

        client_socket.sendall(data) # 回传

        client_socket.close()

 

 

class JSONRPC(object):

    def __init__(self):

        self.data = None

 

    def from_data(self, data):

        '''解析数据'''

        self.data = json.loads(data.decode('utf-8'))

 

    def call_method(self, data):

        '''解析数据，调用对应的方法变将该方法执行结果返回'''

        self.from_data(data)

        method_name = self.data['method_name']

        method_args = self.data['method_args']

        method_kwargs = self.data['method_kwargs']

        res = self.funs[method_name](*method_args, **method_kwargs)

        data = {"res": res}

        return json.dumps(data).encode('utf-8')

 

 

class RPCStub(object):

    def __init__(self):

        self.funs = {}

 

    def register_function(self, function, name=None):

        '''Server端方法注册，Client端只可调用被注册的方法'''

        if name is None:

            name = function.__name__

        self.funs[name] = function

至此，Client端与Server端的代码均编写完毕。

测试环节：