手写 RPC 框架：从原理到实现 ⭐⭐

// 像调用本地方法一样
User user = userService.getUserById(1001);

// 但实际上这是一次网络通信
// 数据经过：序列化 → 网络传输 → 服务端反序列化 → 方法执行
//         → 结果序列化 → 网络传输 → 客户端反序列化 → 返回结果

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;

public class JdkRpcProxy<T> implements InvocationHandler {

    private Class<T> interfaceClass;
    private String serviceVersion;
    private LoadBalancer loadBalancer;
    private List<ServiceNode> serviceNodes;

    /**
     * 创建代理对象
     */
    @SuppressWarnings("unchecked")
    public T createProxy(Class<T> interfaceClass, String serviceVersion,
                        List<ServiceNode> nodes, LoadBalancer loadBalancer) {
        this.interfaceClass = interfaceClass;
        this.serviceVersion = serviceVersion;
        this.serviceNodes = nodes;
        this.loadBalancer = loadBalancer;

        // 使用 JDK 动态代理创建代理对象
        // 任何对 proxy 对象的方法调用都会触发 invoke 方法
        return (T) Proxy.newProxyInstance(
            interfaceClass.getClassLoader(),  // 类加载器
            new Class<?>[]{interfaceClass},    // 代理实现的接口
            this                               // InvocationHandler
        );
    }

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        // 跳过 Object 类的方法（如 toString, hashCode 等）
        if (Object.class.equals(method.getDeclaringClass())) {
            return method.invoke(this, args);
        }

        // 构建 RPC 请求
        RpcRequest request = RpcRequest.builder()
            .interfaceName(interfaceClass.getName())
            .methodName(method.getName())
            .parameterTypes(method.getParameterTypes())
            .parameters(args)
            .serviceVersion(this.serviceVersion)
            .requestId(UUID.randomUUID().toString())
            .timestamp(System.currentTimeMillis())
            .build();

        // 选择服务节点（负载均衡）
        ServiceNode node = loadBalancer.select(serviceNodes);

        // 通过 Netty 发送请求并获取响应
        RpcResponse response = nettyClient.sendRequest(request, node);

        // 返回结果
        if (response.isError()) {
            throw response.getError();
        }
        return response.getResult();
    }
}

import net.sf.cglib.proxy.Enhancer;
import net.sf.cglib.proxy.MethodInterceptor;
import net.sf.cglib.proxy.MethodProxy;

public class CglibRpcProxy implements MethodInterceptor {

    private LoadBalancer loadBalancer;
    private List<ServiceNode> serviceNodes;
    private Class<?> targetClass;

    public Object createProxy(Class<?> targetClass, List<ServiceNode> nodes,
                              LoadBalancer loadBalancer) {
        this.targetClass = targetClass;
        this.serviceNodes = nodes;
        this.loadBalancer = loadBalancer;

        Enhancer enhancer = new Enhancer();
        enhancer.setSuperclass(targetClass);         // 设置父类
        enhancer.setCallback(this);                  // 设置回调
        return enhancer.create();                    // 创建代理对象
    }

    @Override
    public Object intercept(Object obj, Method method, Object[] args,
                            MethodProxy methodProxy) throws Throwable {
        // 构建 RPC 请求（同 JDK 代理）
        RpcRequest request = RpcRequest.builder()
            .interfaceName(targetClass.getName())
            .methodName(method.getName())
            .parameterTypes(method.getParameterTypes())
            .parameters(args)
            .build();

        ServiceNode node = loadBalancer.select(serviceNodes);
        RpcResponse response = nettyClient.sendRequest(request, node);

        if (response.isError()) {
            throw response.getError();
        }
        return response.getResult();
    }
}

import javassist.*;

public class JavassistRpcProxy {

    public static <T> T createProxy(Class<T> interfaceClass,
                                    List<ServiceNode> nodes,
                                    LoadBalancer loadBalancer) throws Exception {
        ClassPool pool = ClassPool.getDefault();
        
        // 创建类名: XxxServiceProxy
        CtClass proxyClass = pool.makeClass(
            interfaceClass.getName() + "Proxy"
        );

        // 添加接口实现
        proxyClass.addInterface(pool.get(interfaceClass.getName()));

        // 添加成员变量
        CtClass nettyClientClass = pool.get(NettyClient.class.getName());
        CtField clientField = new CtField(nettyClientClass, "client", proxyClass);
        proxyClass.addField(clientField);

        // 添加构造方法
        CtConstructor constructor = new CtConstructor(
            new CtClass[]{nettyClientClass}, proxyClass
        );
        constructor.setBody("{ this.client = $1; }");
        proxyClass.addConstructor(constructor);

        // 为每个接口方法生成方法体
        for (CtMethod interfaceMethod : pool.get(interfaceClass.getName()).getMethods()) {
            CtMethod method = new CtMethod(interfaceMethod, proxyClass);
            method.setBody(
                "{ // 构建并发送 RPC 请求\n" +
                "   RpcRequest request = new RpcRequest();\n" +
                "   request.setMethodName(\"" + method.getName() + "\");\n" +
                "   request.setParameters($args);\n" +
                "   RpcResponse response = client.send(request);\n" +
                "   return ($r)response.getResult();\n" +
                "}"
            );
            proxyClass.addMethod(method);
        }

        Class<?> clazz = proxyClass.toClass();
        return interfaceClass.cast(clazz.getConstructor(NettyClient.class)
            .newInstance(nettyClient));
    }
}

public class NettyClient {

    private EventLoopGroup group = new NioEventLoopGroup();
    private Bootstrap bootstrap = new Bootstrap();

    public RpcResponse sendRequest(RpcRequest request, InetSocketAddress address) {
        try {
            Channel channel = bootstrap.connect(address).sync().channel();

            // 创建 RPC 响应处理器，用于异步接收响应
            RpcFuture<RpcResponse> future = new RpcFuture<>();
            pendingRequests.put(request.getRequestId(), future);

            // 发送请求
            channel.writeAndFlush(request);

            // 等待响应（带超时）
            return future.get(3000, TimeUnit.MILLISECONDS);
        } catch (Exception e) {
            throw new RpcException("RPC 调用失败: " + e.getMessage(), e);
        }
    }

    public void init() {
        bootstrap.group(group)
            .channel(NioSocketChannel.class)
            .handler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) {
                    ChannelPipeline pipeline = ch.pipeline();

                    // ========== 编码器 ==========
                    // 1. 先序列化对象为字节数组
                    // 2. 在字节数组前加上 4 字节长度字段
                    pipeline.addLast(new RpcEncoder());

                    // ========== 解码器 ==========
                    // maxFrameLength: 最大帧大小（防止畸形数据）
                    // lengthFieldOffset: 长度字段的偏移量
                    // lengthFieldLength: 长度字段本身占的字节数
                    // lengthAdjustment: 长度调整值
                    // initialBytesToStrip: 解析后从帧头部跳过的字节数
                    pipeline.addLast(new LengthFieldBasedFrameDecoder(
                        65535,     // maxFrameLength: 单帧最大 64KB
                        0,         // lengthFieldOffset: 长度字段在第 0 字节开始
                        4,         // lengthFieldLength: 长度字段占 4 字节
                        0,         // lengthAdjustment: 长度值就是 Body 长度，无需调整
                        4          // initialBytesToStrip: 解码后跳过前 4 字节（长度字段）
                    ));

                    // 解码后的字节数组 → RpcRequest/RpcResponse 对象
                    pipeline.addLast(new RpcDecoder());

                    // 业务处理器：处理接收到的响应
                    pipeline.addLast(new RpcClientHandler());
                }
            })
            .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 3000);
    }
}

public class NettyServer {

    private ServerBootstrap bootstrap = new ServerBootstrap();
    private EventLoopGroup bossGroup = new NioEventLoopGroup(1);
    private EventLoopGroup workerGroup = new NioEventLoopGroup();

    public void bind(int port) {
        bootstrap.group(bossGroup, workerGroup)
            .channel(NioServerSocketChannel.class)
            .childHandler(new ChannelInitializer<SocketChannel>() {
                @Override
                protected void initChannel(SocketChannel ch) {
                    ChannelPipeline pipeline = ch.pipeline();
                    pipeline.addLast(new LengthFieldBasedFrameDecoder(65535, 0, 4, 0, 4));
                    pipeline.addLast(new RpcDecoder());
                    pipeline.addLast(new RpcEncoder());
                    pipeline.addLast(new RpcServerHandler());  // 核心业务处理
                }
            })
            .option(ChannelOption.SO_BACKLOG, 128)
            .childOption(ChannelOption.SO_KEEPALIVE, true);

        ChannelFuture f = bootstrap.bind(port).sync();
        System.out.println("RPC Server started on port " + port);
    }
}

/**
 * 服务端处理器：接收请求、执行业务逻辑、返回响应
 */
public class RpcServerHandler extends ChannelInboundHandlerAdapter {

    @Autowired
    private ServiceRegistry registry;

    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        RpcRequest request = (RpcRequest) msg;
        RpcResponse response = new RpcResponse();
        response.setRequestId(request.getRequestId());

        try {
            // 1. 根据服务名和方法名找到对应的实现
            Object service = registry.getService(request.getInterfaceName());
            if (service == null) {
                throw new RpcException("Service not found: " + request.getInterfaceName());
            }

            // 2. 反射调用方法
            Method method = service.getClass().getMethod(
                request.getMethodName(),
                request.getParameterTypes()
            );
            method.setAccessible(true);
            Object result = method.invoke(service, request.getParameters());

            // 3. 设置返回结果
            response.setResult(result);

        } catch (Exception e) {
            response.setError(e);
        }

        // 4. 发送响应给客户端
        ctx.writeAndFlush(response);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
        cause.printStackTrace();
        ctx.close();
    }
}

public class ZookeeperServiceRegistry implements ServiceRegistry {

    private static final String ZK_ROOT_PATH = "/rpc";
    private ZooKeeper zooKeeper;

    public ZookeeperServiceRegistry(String zkAddress) {
        this.zooKeeper = new ZooKeeper(zkAddress, 5000, watchedEvent -> {
            if (watchedEvent.getType() == Event.EventType.NodeChildrenChanged) {
                // 服务列表变更通知（可用于推送更新）
                System.out.println("服务列表发生变化，重新获取...");
            }
        });
    }

    @Override
    public void register(String serviceName, InetSocketAddress address) {
        try {
            String path = ZK_ROOT_PATH + "/" + serviceName;
            
            // 创建持久化服务节点
            if (zooKeeper.exists(path, false) == null) {
                zooKeeper.create(path, null, 
                    ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            }

            // 创建临时节点（服务实例）
            String instancePath = path + "/" + address.getHostString() + ":" + address.getPort();
            zooKeeper.create(instancePath,
                (serviceName + ":" + System.currentTimeMillis()).getBytes(),
                ZooDefs.Ids.OPEN_ACL_UNSAFE,
                CreateMode.EPHEMERAL  // 关键：临时节点，连接断开自动删除
            );
            System.out.println("服务注册成功: " + instancePath);

        } catch (Exception e) {
            throw new RpcException("ZooKeeper 注册失败", e);
        }
    }

    @Override
    public List<InetSocketAddress> discover(String serviceName) {
        try {
            String path = ZK_ROOT_PATH + "/" + serviceName;
            List<String> children = zooKeeper.getChildren(path, true);

            return children.stream().map(child -> {
                String[] parts = child.split(":");
                return new InetSocketAddress(parts[0], Integer.parseInt(parts[1]));
            }).collect(Collectors.toList());

        } catch (Exception e) {
            return Collections.emptyList();
        }
    }
}

// Nacos 注册示例（使用 nacos-spring-boot-starter）
@NacosInjected
private NamingService namingService;

public void register() throws NacosException {
    // 注册服务：服务名 + IP + 端口 + 集群名
    namingService.registerInstance("user-service", "192.168.1.10", 8080, "DEFAULT_GROUP");
}

public List<Instance> getInstances() throws NacosException {
    // 获取服务所有实例
    return namingService.selectInstances("user-service", true);
}

public class RandomLoadBalancer implements LoadBalancer {

    private final Random random = new Random();

    @Override
    public ServiceNode select(List<ServiceNode> nodes) {
        if (nodes == null || nodes.isEmpty()) {
            throw new RpcException("无可用服务节点");
        }
        int index = random.nextInt(nodes.size());
        return nodes.get(index);
    }
}

public class RoundRobinLoadBalancer implements LoadBalancer {

    private final AtomicInteger counter = new AtomicInteger(0);

    @Override
    public ServiceNode select(List<ServiceNode> nodes) {
        if (nodes == null || nodes.isEmpty()) {
            throw new RpcException("无可用服务节点");
        }
        int index = Math.abs(counter.getAndIncrement()) % nodes.size();
        return nodes.get(index);
    }
}

public class ConsistentHashLoadBalancer implements LoadBalancer {

    private final TreeMap<Long, ServiceNode> virtualNodes = new TreeMap<>();
    private static final int VIRTUAL_NODES = 160;  // 每个物理节点的虚拟节点数

    public ConsistentHashLoadBalancer(List<ServiceNode> nodes) {
        for (ServiceNode node : nodes) {
            addNode(node);
        }
    }

    private void addNode(ServiceNode node) {
        for (int i = 0; i < VIRTUAL_NODES; i++) {
            // 虚拟节点名：物理节点 + 序号
            String virtualNodeName = node.getHost() + ":" + node.getPort() + "-VN" + i;
            long hash = hash(virtualNodeName);
            virtualNodes.put(hash, node);
        }
    }

    private long hash(String key) {
        // 使用 MurmurHash 保证哈希分布均匀
        MurmurHash128 murmurHash = new MurmurHash128();
        long[] h = murmurHash.hash(key);
        return h[0];
    }

    @Override
    public ServiceNode select(List<ServiceNode> nodes, String key) {
        if (virtualNodes.isEmpty()) {
            throw new RpcException("无可用服务节点");
        }
        // 计算 key 的哈希值，查找第一个 >= keyHash 的虚拟节点
        long keyHash = hash(key);
        Map.Entry<Long, ServiceNode> entry = virtualNodes.ceilingEntry(keyHash);
        // 如果 keyHash 大于环上最后一个虚拟节点，则取第一个
        if (entry == null) {
            entry = virtualNodes.firstEntry();
        }
        return entry.getValue();
    }
}

public class NettyRpcClient {

    private static final int DEFAULT_TIMEOUT = 3000; // 默认 3 秒超时

    public RpcResponse sendRequest(RpcRequest request, ServiceNode node) {
        RpcFuture<RpcResponse> future = new RpcFuture<>();
        pendingRequests.put(request.getRequestId(), future);

        channel.writeAndFlush(request);

        try {
            // 等待响应，超时则抛出异常
            return future.get(DEFAULT_TIMEOUT, TimeUnit.MILLISECONDS);
        } catch (TimeoutException e) {
            pendingRequests.remove(request.getRequestId());
            throw new RpcException("RPC 调用超时: " + request.getMethodName());
        }
    }
}

public class RpcClientWithRetry {

    private int maxRetries = 3;
    private long retryDelay = 100; // 重试间隔（毫秒）

    public RpcResponse sendRequestWithRetry(RpcRequest request, List<ServiceNode> nodes,
                                           LoadBalancer balancer) {
        RpcException lastException = null;

        for (int attempt = 0; attempt <= maxRetries; attempt++) {
            try {
                ServiceNode node = balancer.select(nodes);
                return nettyClient.sendRequest(request, node);
            } catch (RpcException e) {
                lastException = e;
                if (attempt < maxRetries) {
                    try {
                        Thread.sleep(retryDelay * (attempt + 1)); // 指数退避
                    } catch (InterruptedException ie) {
                        Thread.currentThread().interrupt();
                    }
                }
            }
        }

        throw new RpcException("RPC 调用重试 " + maxRetries + " 次后仍然失败",
            lastException);
    }
}

// 序列化
UserProto.GetUserRequest request = UserProto.GetUserRequest.newBuilder()
    .setUserId(1001)
    .build();
byte[] bytes = request.toByteArray();

// 反序列化
UserProto.GetUserRequest parsed = UserProto.GetUserRequest.parseFrom(bytes);

public class KryoSerializer implements Serializer {

    private static final ThreadLocal<Kryo> kryo = ThreadLocal.withInitial(() -> {
        Kryo k = new Kryo();
        k.setRegistrationRequired(false);  // 允许未注册类
        k.setReferences(true);
        return k;
    });

    @Override
    public byte[] serialize(Object obj) {
        Kryo k = kryo.get();
        try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
             Output output = new Output(baos)) {
            k.writeObject(output, obj);
            return output.toBytes();
        }
    }

    @Override
    public <T> T deserialize(byte[] bytes, Class<T> clazz) {
        Kryo k = kryo.get();
        try (Input input = new Input(bytes)) {
            return k.readObject(input, clazz);
        }
    }
}

@Data
@Builder
public class RpcRequest implements Serializable {
    private static final long serialVersionUID = 1L;
    
    private String requestId;       // 请求唯一 ID
    private String interfaceName;  // 接口全名
    private String methodName;      // 方法名
    private Class<?>[] parameterTypes; // 参数类型列表
    private Object[] parameters;    // 参数值
    private String serviceVersion; // 服务版本
    private long timestamp;        // 时间戳
}

@Data
@Builder
public class RpcResponse implements Serializable {
    private static final long serialVersionUID = 1L;
    
    private String requestId;       // 对应请求的 ID
    private Object result;          // 返回结果
    private Throwable error;         // 错误信息
    private long duration;          // 耗时
    
    public boolean isError() {
        return error != null;
    }
}

public interface HelloService {
    String sayHello(String name);
    User getUser(Long id);
}

@Data
class User implements Serializable {
    private Long id;
    private String name;
    private String email;
}

@Service
public class HelloServiceImpl implements HelloService {
    @Override
    public String sayHello(String name) {
        return "Hello, " + name + "!";
    }
    
    @Override
    public User getUser(Long id) {
        return new User(id, "User-" + id, "user" + id + "@example.com");
    }
}

public class RpcServerBootstrap {
    public static void main(String[] args) {
        // 创建并启动 RPC 服务端，监听 8080 端口
        RpcServer server = new RpcServer("127.0.0.1", 8080);
        
        // 注册服务
        server.register(HelloService.class, new HelloServiceImpl());
        
        // 启动
        server.start();
        System.out.println("RPC Server started on port 8080");
    }
}

public class RpcClientBootstrap {
    public static void main(String[] args) {
        // 创建 RPC 代理（消费者端）
        RpcClient rpcClient = new RpcClient();
        
        // 通过 JDK 动态代理获取服务代理对象
        HelloService helloService = rpcClient.createProxy(
            HelloService.class,          // 接口类型
            "1.0",                        // 服务版本
            "127.0.0.1:8080"             // 服务地址
        );
        
        // 像调用本地方法一样调用远程服务
        String result = helloService.sayHello("张三");
        System.out.println(result);  // 输出: Hello, 张三!
        
        User user = helloService.getUser(1001L);
        System.out.println(user);     // 输出: User(id=1001, name=User-1001, ...)
    }
}