如何手搓一个自定义的RPC（远程过程调用框架）

//接口方法
ShoppingCart shopping(String pin);

//客户端通过代理工厂实现接口的一个代理实例
IShoppingCartService serviceProxy = ProxyFactory.factory(IShoppingCartService.class)                
                .setSerializerType(SerializerType.JDK) //客户端设置所使用的序列化工具，此处为JDK原生
                .newProxyInstance(); //返回代理 实现

//像调用本地方法一样，调用此代理实例的shopping 方法
ShoppingCart result = serviceProxy.shopping("userPin");
log.info("result={}", JSONObject.toJSONString(result));

public class ProxyFactory {
    //……省略
    
    public I newProxyInstance() {     
        //服务的元数据信息
        ServiceData serviceData = new ServiceData(
                group, //分组
                providerName, //服务名称，一般为接口的class的全限定名称
                StringUtils.isNotBlank(version) ? version : "1.0.0" //版本号
        );


        //调用器
        Calller caller = newCaller().timeoutMillis(timeoutMillis);
        //集群策略，用于实现快速失败或失败转等功能
        Strategy strategy = StrategyConfigContext.of(strategy, retries);
        Object handler = null;
        switch (invokeType) {
            case "syncCall":
                //同步调用handler
                handler = new SyncCaller(serviceData, caller);
                break;
            case "asyncCall":
                //异步调用handler
                handler = new AsyncCaller(client.appName(), serviceData, caller, strategy);
                break;
            default:
                throw new RuntimeException("未知类型: " + invokeType);
        }


        //返回代理实例
        return ProxyEnum.getDefault().newProxy(interfaceClass, handler);
    }
    //……省略
}

//方法newProxy 的具体实现
public  T newProxy(Class interfaceType, Object handler) {
            Class cls = new ByteBuddy()
                     //生成接口的子类
                    .subclass(interfaceType) 
                     //默认代理接口中所有声明的方法
                    .method(ElementMatchers.isDeclaredBy(interfaceType))
                     //代码增强，即接口中所有被代理的方法都
                     //委托给用户自定义的handler处理，这也是动态代理的意义所在
                   .intercept(MethodDelegation.to(handler, "handlerInstance"))
                    .make()
                     //通过类加载器加载
                   .load(interfaceType.getClassLoader(), ClassLoadingStrategy.Default.INJECTION)
                    .getLoaded();


            try {
                //通过newInstance构建一个代理实例并返回    
               return cls.newInstance();
            } catch (Throwable t) {
                ……
            }
        }

public class SyncCaller extends AbstractCaller {
    //……省略   
    
    @RuntimeType
    public Object syncCall(@Origin Method method, @AllArguments @RuntimeType Object[] args) throws Throwable {
        //封装请求的接口中的方法名及方法参数，组成一个request请求对象
        StarGateRequest request = createRequest(methodName, args);
        //集群容错策略调度器接口
        //提供快速失败，失败转移等策略供调用方选择，此处默认采用了快速失败的策略
        Invoker invoker = new FastFailInvoker();
        //returnType 的类型决定了泛型方法的实际结果类型，用于后续调用结果的类型转换
        Future future = invoker.invoke(request, method.getReturnType());
        if (sync) {
            //同步调用，线程会阻塞在get方法，直到超时或结果可用
            Object result = future.getResult();
            return result;
        } else {
            return future;
        }
    }
}


//同步，异步调用的关键点就在于InvokeFuture，它继承了Java的CompletionStage类，用于异步编程

//上述代码行Future future = invoker.invoke(request, method.getReturnType());
//具体实现


public  Future invoke(StarGateRequest request, Class returnType) throws Exception {
        //对象序列化器，默认为JDK
        final Serializer _serializer = serializer();
        //message对象包含此次请求的接口名，方法名及实际参数列表
        final Message message = request.message();
        //通过软负载均衡选择一个 Netty channel
        Channel channel = selectChannel(message.getMetadata());
        byte code = _serializer.code();
        //将message对象序列成字节数组
        byte[] bytes = _serializer.writeObject(message);
        request.bytes(code, bytes);


        //数据写入 channel 并返回 future 约定，用于同步或异步获得调用结果
        return write(channel, request, returnType);
    }

//对象的序列化，JDK 原生方式
public  byte[] writeObject(T obj) {
        ByteArrayOutputStream buf = OutputStreams.getByteArrayOutputStream();
        try (ObjectOutputStream output = new ObjectOutputStream(buf)) {
            output.writeObject(obj);
            output.flush();
            return buf.toByteArray();
        } catch (IOException e) {
            ThrowUtil.throwException(e);
        } finally {
            OutputStreams.resetBuf(buf);
        }
        return null; 
    }

//上述代码  write(channel, request, returnType);
//具体实现
protected  DefaultFuture write(final Channel channel,
                                               final StarGateRequest request,
                                               final Class returnType) {
        //……省略


        //调用结果占位 future对象，这也是promise编程模式
        final Future future = DefaultFuture.newFuture(request.invokeId(), channel, timeoutMillis, returnType);


        //将请求负载对象写入Netty channel通道，并绑定监听器处理写入结果
        channel.writeAndFlush(request).addListener((ChannelFutureListener) listener -> {
            if (listener.isSuccess()) {
                //网络写入成功
                ……
            } else {
                //异常时，构造造调用结果，供调用方进行处理
                DefaultFuture.errorFuture(channel, response, dispatchType);
            }
        });


        //因为Netty 是非阻塞的，所以写入后可立刻返回
        return future;
    }

//自定义的编码器 继承自Netty 的 MessageToByteEncoder
public class StarGateEncoder extends MessageToByteEncoder {


    //……省略


    private void doEncodeRequest(RequestPayload request, ByteBuf out) {
        byte sign = StarGateProtocolHeader.toSign(request.serializerCode(), StarGateProtocolHeader.REQUEST);
        long invokeId = request.invokeId();
        byte[] bytes = request.bytes();
        int length = bytes.length;


        out.writeShort(StarGateProtocolHeader.Head)  //写入两个字节
                .writeByte(sign)              //写入1个字节
                .writeByte(0x00)            //写入1个字节
                .writeLong(invokeId)          //写入8个节节
                .writeInt(length)             //写入4个字节
                .writeBytes(bytes);
    }


}

//消息解码器，继承自Netty 的ReplayingDecoder，将客户端请求解码为 RequestPayload 对象，供业务处理handler使用
public class StarGateDecoder extends ReplayingDecoder {


    //……省略


    @Override
    protected void decode(ChannelHandlerContext ctx, ByteBuf in, List