跳至主要內容

RPC 实现基础

老马啸西风2025/8/30大约 10 分钟rpcrpc

在深入学习 RPC（Remote Procedure Call）框架之前，理解其底层实现原理和基础技术是至关重要的。RPC 作为一种分布式通信技术，涉及网络编程、序列化、代理模式、反射机制等多个计算机科学领域的核心概念。本章将从基础开始，逐步深入探讨 RPC 实现所需的核心技术，为后续章节中从零实现 RPC 框架奠定坚实的理论基础。

RPC 的基本工作原理

什么是 RPC

RPC 是一种进程间通信方式，它允许程序调用另一个地址空间（通常是网络上的另一台机器）的过程或函数，就像调用本地函数一样。这种透明性极大地简化了分布式系统的开发复杂性。

RPC 的核心组件

一个典型的 RPC 系统包含以下几个核心组件：

  1. 客户端（Client）：发起远程调用的程序
  2. 客户端存根（Client Stub）：负责将调用参数序列化并发送给服务端
  3. 服务端存根（Server Stub）：接收客户端请求，将参数反序列化并调用实际的服务方法
  4. 服务端（Server）：提供远程服务的程序
  5. 网络传输：负责在客户端和服务端之间传输数据

RPC 调用流程

RPC 的完整调用流程如下：

  1. 客户端调用客户端存根（Client Stub）
  2. 客户端存根将参数打包成消息（参数序列化）
  3. 客户端存根通过网络将消息发送到服务端
  4. 服务端存根接收网络消息
  5. 服务端存根将消息解包（参数反序列化）
  6. 服务端存根调用服务端程序
  7. 服务端程序执行并将结果返回给服务端存根
  8. 服务端存根将结果打包成消息（结果序列化）
  9. 服务端存根通过网络将消息发送回客户端
  10. 客户端存根接收网络消息
  11. 客户端存根将消息解包（结果反序列化）
  12. 客户端存根将结果返回给客户端

网络编程基础

Socket 编程

Socket 是网络编程的基础，RPC 框架通常基于 Socket 实现网络通信：

// 基础的 Socket 服务端示例
public class SimpleSocketServer {
    private int port;
    
    public SimpleSocketServer(int port) {
        this.port = port;
    }
    
    public void start() throws IOException {
        ServerSocket serverSocket = new ServerSocket(port);
        System.out.println("Server started on port " + port);
        
        while (true) {
            Socket clientSocket = serverSocket.accept();
            // 为每个客户端连接创建一个处理线程
            new Thread(new ClientHandler(clientSocket)).start();
        }
    }
    
    private static class ClientHandler implements Runnable {
        private Socket socket;
        
        public ClientHandler(Socket socket) {
            this.socket = socket;
        }
        
        @Override
        public void run() {
            try (BufferedReader in = new BufferedReader(new InputStreamReader(socket.getInputStream()));
                 PrintWriter out = new PrintWriter(socket.getOutputStream(), true)) {
                
                String inputLine;
                while ((inputLine = in.readLine()) != null) {
                    System.out.println("Received: " + inputLine);
                    // 回显消息
                    out.println("Echo: " + inputLine);
                }
            } catch (IOException e) {
                e.printStackTrace();
            } finally {
                try {
                    socket.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
    }
}

// 基础的 Socket 客户端示例
public class SimpleSocketClient {
    private String host;
    private int port;
    
    public SimpleSocketClient(String host, int port) {
        this.host = host;
        this.port = port;
    }
    
    public void sendMessage(String message) throws IOException {
        try (Socket socket = new Socket(host, port);
             PrintWriter out = new PrintWriter(socket.getOutputStream(), true);
             BufferedReader in = new BufferedReader(new InputStreamReader(socket.getInputStream()))) {
            
            // 发送消息
            out.println(message);
            
            // 接收响应
            String response = in.readLine();
            System.out.println("Server response: " + response);
        }
    }
}

NIO（Non-blocking I/O）

现代 RPC 框架通常使用 NIO 来提高并发处理能力：

// 基础的 NIO 服务端示例
public class SimpleNioServer {
    private Selector selector;
    private ServerSocketChannel serverChannel;
    private int port;
    
    public SimpleNioServer(int port) throws IOException {
        this.port = port;
        init();
    }
    
    private void init() throws IOException {
        selector = Selector.open();
        serverChannel = ServerSocketChannel.open();
        serverChannel.configureBlocking(false);
        serverChannel.socket().bind(new InetSocketAddress(port));
        serverChannel.register(selector, SelectionKey.OP_ACCEPT);
        
        System.out.println("NIO Server started on port " + port);
    }
    
    public void start() throws IOException {
        while (true) {
            selector.select();
            Set<SelectionKey> selectedKeys = selector.selectedKeys();
            Iterator<SelectionKey> iter = selectedKeys.iterator();
            
            while (iter.hasNext()) {
                SelectionKey key = iter.next();
                
                if (key.isAcceptable()) {
                    handleAccept(key);
                }
                
                if (key.isReadable()) {
                    handleRead(key);
                }
                
                iter.remove();
            }
        }
    }
    
    private void handleAccept(SelectionKey key) throws IOException {
        ServerSocketChannel serverChannel = (ServerSocketChannel) key.channel();
        SocketChannel clientChannel = serverChannel.accept();
        clientChannel.configureBlocking(false);
        clientChannel.register(selector, SelectionKey.OP_READ);
        System.out.println("Client connected: " + clientChannel.getRemoteAddress());
    }
    
    private void handleRead(SelectionKey key) throws IOException {
        SocketChannel clientChannel = (SocketChannel) key.channel();
        ByteBuffer buffer = ByteBuffer.allocate(1024);
        
        int bytesRead = clientChannel.read(buffer);
        if (bytesRead == -1) {
            clientChannel.close();
            return;
        }
        
        buffer.flip();
        byte[] data = new byte[buffer.remaining()];
        buffer.get(data);
        String message = new String(data).trim();
        
        System.out.println("Received: " + message);
        
        // 回显消息
        String response = "Echo: " + message;
        ByteBuffer responseBuffer = ByteBuffer.wrap(response.getBytes());
        clientChannel.write(responseBuffer);
    }
}

序列化技术

Java 原生序列化

Java 原生序列化是最简单的序列化方式：

// Java 原生序列化示例
public class JavaSerializationExample {
    // 可序列化的数据类
    public static class Person implements Serializable {
        private static final long serialVersionUID = 1L;
        
        private String name;
        private int age;
        private String email;
        
        public Person(String name, int age, String email) {
            this.name = name;
            this.age = age;
            this.email = email;
        }
        
        // getter 和 setter 方法
        public String getName() { return name; }
        public void setName(String name) { this.name = name; }
        
        public int getAge() { return age; }
        public void setAge(int age) { this.age = age; }
        
        public String getEmail() { return email; }
        public void setEmail(String email) { this.email = email; }
        
        @Override
        public String toString() {
            return "Person{name='" + name + "', age=" + age + ", email='" + email + "'}";
        }
    }
    
    // 序列化方法
    public static byte[] serialize(Object obj) throws IOException {
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        ObjectOutputStream oos = new ObjectOutputStream(bos);
        oos.writeObject(obj);
        oos.close();
        return bos.toByteArray();
    }
    
    // 反序列化方法
    public static Object deserialize(byte[] data) throws IOException, ClassNotFoundException {
        ByteArrayInputStream bis = new ByteArrayInputStream(data);
        ObjectInputStream ois = new ObjectInputStream(bis);
        Object obj = ois.readObject();
        ois.close();
        return obj;
    }
    
    // 测试示例
    public static void main(String[] args) {
        try {
            // 创建对象
            Person person = new Person("John Doe", 30, "john.doe@example.com");
            
            // 序列化
            byte[] serializedData = serialize(person);
            System.out.println("Serialized data length: " + serializedData.length);
            
            // 反序列化
            Person deserializedPerson = (Person) deserialize(serializedData);
            System.out.println("Deserialized person: " + deserializedPerson);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

JSON 序列化

JSON 是一种轻量级的数据交换格式：

// JSON 序列化示例（使用 Jackson）
public class JsonSerializationExample {
    private static final ObjectMapper mapper = new ObjectMapper();
    
    // 序列化方法
    public static String serialize(Object obj) throws JsonProcessingException {
        return mapper.writeValueAsString(obj);
    }
    
    // 反序列化方法
    public static <T> T deserialize(String json, Class<T> clazz) throws JsonProcessingException {
        return mapper.readValue(json, clazz);
    }
    
    // 测试示例
    public static void main(String[] args) {
        try {
            // 创建对象
            Person person = new Person("John Doe", 30, "john.doe@example.com");
            
            // 序列化
            String jsonString = serialize(person);
            System.out.println("JSON string: " + jsonString);
            System.out.println("JSON string length: " + jsonString.length());
            
            // 反序列化
            Person deserializedPerson = deserialize(jsonString, Person.class);
            System.out.println("Deserialized person: " + deserializedPerson);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

Protocol Buffers

Protocol Buffers 是 Google 开发的高效序列化框架：

// Protocol Buffers 序列化示例
// 首先定义 .proto 文件
// person.proto
/*
syntax = "proto3";

message Person {
    string name = 1;
    int32 age = 2;
    string email = 3;
}
*/

public class ProtobufSerializationExample {
    // 序列化方法
    public static byte[] serialize(Person person) {
        return person.toByteArray();
    }
    
    // 反序列化方法
    public static Person deserialize(byte[] data) throws InvalidProtocolBufferException {
        return Person.parseFrom(data);
    }
    
    // 测试示例
    public static void main(String[] args) {
        try {
            // 创建对象
            Person.Builder builder = Person.newBuilder();
            builder.setName("John Doe");
            builder.setAge(30);
            builder.setEmail("john.doe@example.com");
            Person person = builder.build();
            
            // 序列化
            byte[] serializedData = serialize(person);
            System.out.println("Serialized data length: " + serializedData.length);
            
            // 反序列化
            Person deserializedPerson = deserialize(serializedData);
            System.out.println("Deserialized person: " + deserializedPerson);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

动态代理技术

JDK 动态代理

JDK 动态代理是实现 RPC 客户端代理的核心技术：

// JDK 动态代理示例
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;

// 服务接口
public interface UserService {
    User getUserById(String userId);
    List<User> getAllUsers();
}

// 服务实现
public class UserServiceImpl implements UserService {
    @Override
    public User getUserById(String userId) {
        return new User(userId, "John Doe", "john.doe@example.com");
    }
    
    @Override
    public List<User> getAllUsers() {
        List<User> users = new ArrayList<>();
        users.add(new User("1", "John Doe", "john.doe@example.com"));
        users.add(new User("2", "Jane Smith", "jane.smith@example.com"));
        return users;
    }
}

// 调用处理器
public class RpcInvocationHandler implements InvocationHandler {
    private String serviceName;
    private Class<?> serviceInterface;
    
    public RpcInvocationHandler(String serviceName, Class<?> serviceInterface) {
        this.serviceName = serviceName;
        this.serviceInterface = serviceInterface;
    }
    
    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        // 构建 RPC 请求
        RpcRequest request = new RpcRequest();
        request.setServiceName(serviceName);
        request.setMethodName(method.getName());
        request.setParameterTypes(method.getParameterTypes());
        request.setParameters(args);
        
        // 发送请求并获取响应（简化处理）
        RpcResponse response = sendRequest(request);
        
        // 处理响应
        if (response.hasError()) {
            throw response.getError();
        }
        
        return response.getData();
    }
    
    private RpcResponse sendRequest(RpcRequest request) {
        // 实现网络请求发送逻辑
        // 这里简化处理，实际实现会涉及网络通信、序列化等
        return new RpcResponse();
    }
    
    // 创建代理对象的工厂方法
    public static <T> T createProxy(Class<T> serviceInterface, String serviceName) {
        return (T) Proxy.newProxyInstance(
            serviceInterface.getClassLoader(),
            new Class[]{serviceInterface},
            new RpcInvocationHandler(serviceName, serviceInterface)
        );
    }
}

// 测试示例
public class DynamicProxyTest {
    public static void main(String[] args) {
        // 创建代理对象
        UserService userService = RpcInvocationHandler.createProxy(UserService.class, "userService");
        
        // 调用方法（实际会通过代理处理）
        User user = userService.getUserById("1");
        System.out.println("User: " + user);
        
        List<User> users = userService.getAllUsers();
        System.out.println("Users: " + users);
    }
}

CGLIB 动态代理

CGLIB 动态代理可以代理没有实现接口的类：

// CGLIB 动态代理示例
import net.sf.cglib.proxy.Enhancer;
import net.sf.cglib.proxy.MethodInterceptor;
import net.sf.cglib.proxy.MethodProxy;

public class CglibRpcInterceptor implements MethodInterceptor {
    private String serviceName;
    
    public CglibRpcInterceptor(String serviceName) {
        this.serviceName = serviceName;
    }
    
    @Override
    public Object intercept(Object obj, Method method, Object[] args, MethodProxy proxy) throws Throwable {
        // 构建 RPC 请求
        RpcRequest request = new RpcRequest();
        request.setServiceName(serviceName);
        request.setMethodName(method.getName());
        request.setParameterTypes(method.getParameterTypes());
        request.setParameters(args);
        
        // 发送请求并获取响应
        RpcResponse response = sendRequest(request);
        
        // 处理响应
        if (response.hasError()) {
            throw response.getError();
        }
        
        return response.getData();
    }
    
    private RpcResponse sendRequest(RpcRequest request) {
        // 实现网络请求发送逻辑
        return new RpcResponse();
    }
    
    // 创建代理对象的工厂方法
    public static <T> T createProxy(Class<T> clazz, String serviceName) {
        Enhancer enhancer = new Enhancer();
        enhancer.setSuperclass(clazz);
        enhancer.setCallback(new CglibRpcInterceptor(serviceName));
        return (T) enhancer.create();
    }
}

反射机制

Java 反射基础

反射机制是实现 RPC 服务端调用的核心技术：

// 反射机制示例
public class ReflectionExample {
    // 目标类
    public static class Calculator {
        public int add(int a, int b) {
            return a + b;
        }
        
        public int multiply(int a, int b) {
            return a * b;
        }
        
        private void privateMethod() {
            System.out.println("This is a private method");
        }
    }
    
    public static void main(String[] args) {
        try {
            // 获取类信息
            Class<?> clazz = Calculator.class;
            System.out.println("Class name: " + clazz.getName());
            
            // 创建实例
            Object instance = clazz.newInstance();
            
            // 获取方法信息
            Method[] methods = clazz.getMethods();
            System.out.println("Public methods:");
            for (Method method : methods) {
                System.out.println("  " + method.getName());
            }
            
            // 调用方法
            Method addMethod = clazz.getMethod("add", int.class, int.class);
            Object result = addMethod.invoke(instance, 5, 3);
            System.out.println("5 + 3 = " + result);
            
            // 调用私有方法
            Method privateMethod = clazz.getDeclaredMethod("privateMethod");
            privateMethod.setAccessible(true); // 设置可访问
            privateMethod.invoke(instance);
            
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

在 RPC 中的应用

// RPC 中反射的应用
public class RpcServiceInvoker {
    private Map<String, Object> serviceMap = new ConcurrentHashMap<>();
    
    public void registerService(String serviceName, Object serviceImpl) {
        serviceMap.put(serviceName, serviceImpl);
    }
    
    public RpcResponse invoke(RpcRequest request) {
        RpcResponse response = new RpcResponse();
        try {
            // 查找服务实现
            Object serviceImpl = serviceMap.get(request.getServiceName());
            if (serviceImpl == null) {
                throw new RuntimeException("Service not found: " + request.getServiceName());
            }
            
            // 获取类信息
            Class<?> serviceClass = serviceImpl.getClass();
            
            // 获取方法
            Method method = serviceClass.getMethod(request.getMethodName(), request.getParameterTypes());
            
            // 调用方法
            Object result = method.invoke(serviceImpl, request.getParameters());
            
            // 设置响应结果
            response.setData(result);
        } catch (Exception e) {
            response.setError(e);
        }
        return response;
    }
}

网络协议设计

自定义协议格式

设计一个简单的 RPC 协议格式：

// 自定义协议格式
// +--------+--------+--------+--------+--------+--------+--------+--------+
// |  魔数  | 主版本 | 次版本 |  操作  |        序列化方式        |  数据长度  |
// +--------+--------+--------+--------+--------+--------+--------+--------+
// |                              数据内容                               |
// +--------+--------+--------+--------+--------+--------+--------+--------+

public class CustomProtocol {
    // 协议常量
    public static final int MAGIC_NUMBER = 0x12345678;
    public static final byte VERSION_MAJOR = 1;
    public static final byte VERSION_MINOR = 0;
    
    // 操作类型
    public static final byte OPERATION_REQUEST = 1;
    public static final byte OPERATION_RESPONSE = 2;
    
    // 序列化方式
    public static final byte SERIALIZATION_JAVA = 1;
    public static final byte SERIALIZATION_JSON = 2;
    public static final byte SERIALIZATION_PROTOBUF = 3;
    
    // 编码请求
    public static byte[] encodeRequest(RpcRequest request, byte serializationType) throws Exception {
        ByteArrayOutputStream bos = new ByteArrayOutputStream();
        DataOutputStream dos = new DataOutputStream(bos);
        
        // 写入协议头
        dos.writeInt(MAGIC_NUMBER);
        dos.writeByte(VERSION_MAJOR);
        dos.writeByte(VERSION_MINOR);
        dos.writeByte(OPERATION_REQUEST);
        dos.writeByte(serializationType);
        
        // 序列化数据
        byte[] data;
        switch (serializationType) {
            case SERIALIZATION_JAVA:
                data = JavaSerializationExample.serialize(request);
                break;
            case SERIALIZATION_JSON:
                data = JsonSerializationExample.serialize(request).getBytes();
                break;
            case SERIALIZATION_PROTOBUF:
                // 假设 request 有 toByteArray 方法
                data = request.toByteArray();
                break;
            default:
                throw new IllegalArgumentException("Unsupported serialization type: " + serializationType);
        }
        
        // 写入数据长度和数据
        dos.writeInt(data.length);
        dos.write(data);
        
        dos.close();
        return bos.toByteArray();
    }
    
    // 解码请求
    public static RpcRequest decodeRequest(byte[] data) throws Exception {
        ByteArrayInputStream bis = new ByteArrayInputStream(data);
        DataInputStream dis = new DataInputStream(bis);
        
        // 读取协议头
        int magicNumber = dis.readInt();
        if (magicNumber != MAGIC_NUMBER) {
            throw new IllegalArgumentException("Invalid magic number");
        }
        
        byte versionMajor = dis.readByte();
        byte versionMinor = dis.readByte();
        byte operation = dis.readByte();
        byte serializationType = dis.readByte();
        int dataLength = dis.readInt();
        
        // 读取数据
        byte[] requestData = new byte[dataLength];
        dis.readFully(requestData);
        
        // 反序列化数据
        switch (serializationType) {
            case SERIALIZATION_JAVA:
                return (RpcRequest) JavaSerializationExample.deserialize(requestData);
            case SERIALIZATION_JSON:
                return JsonSerializationExample.deserialize(new String(requestData), RpcRequest.class);
            case SERIALIZATION_PROTOBUF:
                // 假设 RpcRequest 有 parseFrom 方法
                return RpcRequest.parseFrom(requestData);
            default:
                throw new IllegalArgumentException("Unsupported serialization type: " + serializationType);
        }
    }
}

连接管理

连接池实现

连接池是提高 RPC 性能的重要技术：

// 连接池实现
public class ConnectionPool {
    private Queue<Socket> connectionPool = new ConcurrentLinkedQueue<>();
    private String host;
    private int port;
    private int maxPoolSize;
    private long maxIdleTime; // 最大空闲时间
    
    public ConnectionPool(String host, int port, int maxPoolSize, long maxIdleTime) {
        this.host = host;
        this.port = port;
        this.maxPoolSize = maxPoolSize;
        this.maxIdleTime = maxIdleTime;
    }
    
    public Socket getConnection() throws IOException {
        Socket socket = connectionPool.poll();
        if (socket == null || socket.isClosed() || isExpired(socket)) {
            // 创建新连接
            socket = new Socket(host, port);
            socket.setKeepAlive(true); // 启用 TCP keep-alive
        }
        return socket;
    }
    
    public void releaseConnection(Socket socket) {
        if (connectionPool.size() < maxPoolSize && !socket.isClosed() && !isExpired(socket)) {
            connectionPool.offer(socket);
        } else {
            try {
                socket.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
    
    private boolean isExpired(Socket socket) {
        // 检查连接是否过期
        return System.currentTimeMillis() - socket.getLastActivityTime() > maxIdleTime;
    }
}

容错机制

超时控制

// 超时控制实现
public class TimeoutHandler {
    public <T> T executeWithTimeout(Callable<T> task, long timeout, TimeUnit unit) 
            throws TimeoutException, ExecutionException, InterruptedException {
        ExecutorService executor = Executors.newSingleThreadExecutor();
        try {
            Future<T> future = executor.submit(task);
            return future.get(timeout, unit);
        } finally {
            executor.shutdown();
        }
    }
}

重试机制

// 重试机制实现
public class RetryHandler {
    public <T> T executeWithRetry(Callable<T> task, int maxRetries) 
            throws Exception {
        Exception lastException = null;
        
        for (int i = 0; i <= maxRetries; i++) {
            try {
                return task.call();
            } catch (Exception e) {
                lastException = e;
                if (i < maxRetries) {
                    // 等待后重试
                    Thread.sleep(1000 * (i + 1)); // 指数退避
                }
            }
        }
        
        throw lastException;
    }
}

总结

本章介绍了实现 RPC 框架所需的基础技术，包括网络编程、序列化、动态代理、反射机制、协议设计、连接管理和容错机制等。这些技术是构建高性能、可靠的 RPC 框架的基石。

通过本章的学习，我们应该能够：

  1. 理解 RPC 的基本工作原理和核心组件
  2. 掌握 Socket 和 NIO 网络编程技术
  3. 了解不同的序列化方式及其特点
  4. 理解动态代理和反射机制的原理和应用
  5. 掌握自定义协议设计的基本方法
  6. 了解连接池和容错机制的实现

在后续章节中，我们将基于这些基础技术，从零开始实现一个完整的 RPC 框架，进一步加深对 RPC 技术的理解。

我是老马，期待与你的下次重逢
Copyright © 2025 老马啸西风