Triển khai RPC đơn giản trong java

public class NettyTransport {
//..........
    public void server() throws Exception {
        this.group = new NioEventLoopGroup();
        bootstrap = new ServerBootstrap();
        bootstrap.group(group)
                .channel(NioServerSocketChannel.class)
                .option(ChannelOption.TCP_NODELAY,true)
                .handler(new LoggingHandler(LogLevel.INFO))
                .childHandler(new ChannelInitializer() {
                    protected void initChannel(Channel channel) throws Exception {
                        channel.pipeline().addLast(new RpcHandle(listener));
                    }
                });
        ChannelFuture f = bootstrap.bind(port).sync(); // (5)
    }
}

    public void connectNetty() {
        EventLoopGroup workerGroup = new NioEventLoopGroup();

        try {
            Bootstrap b = new Bootstrap(); // (1)
            b.group(workerGroup); // (2)
            b.channel(NioSocketChannel.class); // (3)
            b.option(ChannelOption.SO_KEEPALIVE, true); // (4)
            b.option(ChannelOption.TCP_NODELAY,true);
            b.handler(new ChannelInitializer<SocketChannel>() {
                @Override
                public void initChannel(SocketChannel ch) throws Exception {
                    ch.pipeline().addLast(transport = new ClientHandle(SimpleRpcClient.this));
                }
            });
            // Start the client.
            ChannelFuture f = b.connect(host, port).sync(); // (5)
            // Wait until the connection is closed.
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

import com.esotericsoftware.kryo.kryo5.Kryo;
import com.esotericsoftware.kryo.kryo5.io.ByteBufferInput;
import com.esotericsoftware.kryo.kryo5.io.ByteBufferOutput;
import com.esotericsoftware.kryo.kryo5.io.Input;
import com.esotericsoftware.kryo.kryo5.io.Output;

public class RPCSerialize {
    public static RPCSerialize serialize = new RPCSerialize();

    private final Kryo kryo;

    public RPCSerialize() {
        this.kryo = new Kryo();
        kryo.register(RPCRequest.class);
        kryo.register(RPCResponse.class);
        kryo.register(Object[].class);
        kryo.register(User.class);
    }

    public <T> T deSerialize(byte[] arr, Class<T> t) {
        Input input = new ByteBufferInput(arr);
        return this.kryo.readObject(input, t);
    }

    public <T> byte[] serialize(T t) {
        Output output = new ByteBufferOutput(1024);
        kryo.writeObject(output, t);
        return output.toBytes();
    }
}

import blog.rpc.support.User;

public interface SimpleService {

    int add(int a, int b);

    User getUser();

}

import blog.rpc.service.SimpleService;
import blog.rpc.support.User;

public class ServerImpl implements SimpleService {
    @Override
    public int add(int a, int b) {
        return a+b;
    }

    @Override
    public User getUser() {
        return new User("demtv");
    }
}

import blog.rpc.listener.RpcListener;
import blog.rpc.service.SimpleService;
import blog.rpc.service.impl.ServerImpl;
import blog.rpc.support.*;
import blog.rpc.tranport.imp.NettyTransport;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;

import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.Map;

public class SimpleRpcServer implements RpcListener<ByteBuf> {

    private Map<Integer, XMethod> methodService;

    private RPCSerialize serialize = RPCSerialize.serialize;

    public SimpleRpcServer(int port) {
        new NettyTransport(this, port);
        this.methodService = new HashMap<>();
    }

    public void addHandle(Object target) {
        Method[] methods = target.getClass().getDeclaredMethods();
        for (Method method : methods) {
            XMethod xMethod = new XMethod(method, target);
            methodService.put(xMethod.getSignature(), xMethod);
        }
    }

    public XFuture<ByteBuf> onMessage(ByteBuf buf) {
        byte[] bytes;
        int length = buf.readableBytes();
        if (buf.hasArray()) {
            bytes = buf.array();
        } else {
            bytes = new byte[length];
            buf.getBytes(buf.readerIndex(), bytes);
        }

        RPCRequest request = serialize.deSerialize(bytes, RPCRequest.class);
        try {
            Object result = this.methodService.get(request.getMethod()).invoke(request.getArgs());
            RPCResponse response = new RPCResponse(request.getId(), result);
            XFuture<ByteBuf> future = new XFuture<>();
            byte[]x = serialize.serialize(response);
            future.setResult(Unpooled.copiedBuffer(x));
            return future;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }
}

    @Override
    public int add(int a, int b) {
        int rs = Integer.MAX_VALUE;
        try {
            RPCRequest request = new RPCRequest();
            request.setId(idGenerator.nextId());
            Method method = this.getClass().getMethod("add", int.class, int.class);
            request.setMethod(XMethod.signature(method));
            request.setArgs(new Object[]{a, b});
            byte[] x = serialize.serialize(request);
            transport.sendMsg(Unpooled.copiedBuffer(x));
            XFuture<Integer> future = new XFuture<>();
            futures.put(request.getId(), future);
            rs = future.get(100, TimeUnit.SECONDS);

        } catch (Exception e) {
            e.printStackTrace();
        }
        return rs;
    }

    @Override
    public User getUser() {
        try {
            RPCRequest request = new RPCRequest();
            request.setId(idGenerator.nextId());
            Method method = this.getClass().getMethod("getUser");
            request.setMethod(XMethod.signature(method));
            request.setArgs(null);
            byte[] x = serialize.serialize(request);
            transport.sendMsg(Unpooled.copiedBuffer(x));
            XFuture<User> future = new XFuture<>();
            futures.put(request.getId(), future);
            return future.get(100, TimeUnit.SECONDS);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }