手寫rpc實現詳解

一、手寫rpc框架

RPC（Remote Procedure Call，遠程過程調用），是一種用於客戶端和伺服器端進行通信的協議。通過RPC可以實現跨語言、跨平台、不同機器之間的服務調用，讓遠程的伺服器和本地的客戶端就像本地調用一樣方便地進行通信。手寫rpc框架的實現，可以讓我們更加深入地了解RPC背後的原理，同時也能夠提高我們的編程能力。

下面，我們將從手寫rpc框架的搭建和實現兩個方面，展開對手寫rpc的詳細闡述。

二、手寫rpc框架搭建

1、搭建服務端

在搭建服務端時，我們需要考慮如何處理請求、如何暴露服務、如何啟動服務等問題。

private Executor executor = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());

public void start() throws IOException {
    ServerSocket serverSocket = new ServerSocket(port);
    while (true) {
        Socket socket = serverSocket.accept();
        executor.execute(new RequestHandlerRunnable(socket, service));
    }
}

private static class RequestHandlerRunnable implements Runnable {
    private final Socket socket;
    private final Object service;

    private RequestHandlerRunnable(Socket socket, Object service) {
        this.socket = socket;
        this.service = service;
    }

    @Override
    public void run() {
        try {
            ObjectInputStream input = new ObjectInputStream(socket.getInputStream());
            try {
                String methodName = input.readUTF();
                Class[] parameterTypes = (Class[]) input.readObject();
                Object[] arguments = (Object[]) input.readObject();

                Method method = service.getClass().getMethod(methodName, parameterTypes);
                Object result = method.invoke(service, arguments);

                ObjectOutputStream output = new ObjectOutputStream(socket.getOutputStream());
                output.writeObject(result);
            } catch (Throwable t) {
                ObjectOutputStream output = new ObjectOutputStream(socket.getOutputStream());
                output.writeObject(t);
            } finally {
                input.close();
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
}

2、搭建客戶端

在搭建客戶端時，我們需要考慮如何處理請求、如何發送請求、如何接收返回值等問題。

public class RpcClient {
    private final String host;
    private final int port;

    public RpcClient(String host, int port) {
        this.host = host;
        this.port = port;
    }

    public Object call(String methodName, Class[] parameterTypes, Object[] arguments) throws Throwable {
        Socket socket = new Socket(host, port);
        try {
            ObjectOutputStream output = new ObjectOutputStream(socket.getOutputStream());
            output.writeUTF(methodName);
            output.writeObject(parameterTypes);
            output.writeObject(arguments);

            ObjectInputStream input = new ObjectInputStream(socket.getInputStream());
            Object result = input.readObject();

            if (result instanceof Throwable) {
                throw (Throwable) result;
            }
            return result;
        } finally {
            socket.close();
        }
    }
}

三、手寫rpc框架實現

1、服務註冊與發現

服務註冊和發現是指將提供的服務註冊到註冊中心，並通過查詢註冊中心獲取已註冊服務的IP和埠，從而實現客戶端的遠程調用。

public interface ServiceRegistry {
    /**
     * Register a service to the registry with its host and port
     *
     * @param serviceName  the name of service to register
     * @param serviceHost  the host of the service provider
     * @param servicePort  the port of the service provider
     */
    void register(String serviceName, String serviceHost, int servicePort) throws Exception;

    /**
     * Lookup for available service providers of the {@code serviceName}
     *
     * @param serviceName the name of service that given the service providers provides
     * @return a list of all service providers' {@link ServiceEndPoint} of the given service name
     */
    List lookup(String serviceName) throws Exception;
}

2、動態代理和序列化

客戶端利用動態代理，將需要遠程調用的介面動態生成代碼，實現遠程介面的本地代理。在本地代理方法中，需要將介面、方法、參數等信息序列化並發送到服務端，同時需要接收服務端返回的數據，並進行反序列化，將結果返回給調用方。

public class RpcProxy implements InvocationHandler, Serializable {
    private static final long serialVersionUID = 9130440123381758788L;
    private final Class interfaceClazz;
    private final String serviceName;

    public RpcProxy(Class interfaceClazz, String serviceName) {
        this.interfaceClazz = interfaceClazz;
        this.serviceName = serviceName;
    }

    @SuppressWarnings("unchecked")
    public T getProxy() {
        return (T) Proxy.newProxyInstance(interfaceClazz.getClassLoader(), new Class[] {interfaceClazz}, this);
    }

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        ServiceDiscovery serviceDiscovery = new ServiceDiscoveryImpl();
        InetSocketAddress serviceAddress = serviceDiscovery.lookup(serviceName);

        try {
            RpcClient rpcClient = new RpcClient(serviceAddress.getAddress().getHostAddress(), serviceAddress.getPort());
            RpcRequest request = new RpcRequest();
            request.setServiceName(serviceName);
            request.setMethodName(method.getName());
            request.setParameterTypes(method.getParameterTypes());
            request.setArguments(args);

            return rpcClient.call(request);
        } catch (Exception e) {
            e.printStackTrace();
            return null;
        }
    }
}

public class RpcDecoder extends ByteToMessageDecoder {
    private Class genericClass;

    public RpcDecoder(Class genericClass) {
        this.genericClass = genericClass;
    }

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