深入理解远程过程调用

清疚 2023-05-21 07:03 3阅读 0赞

### 文章目录 ###

*   *   *  一、什么是远程过程调用
         *   *  1.1、一体化调用
             *  1.2、远程过程调用
         *  二、用Java实现简单RPC
         *   *  2.1、业务侵入型Socket
             *  2.2、无业务侵入型Socket
             *  2.3、动态代理无业务侵入型Socket
             *  2.4、Hessian序列化协议
             *  2.5、RMI远程方法调用协议
             *  2.6、Dubbo服务治理框架

> 本文主要参考自相关书籍和网络文章，并附上自身的一些理解，如有遗漏或错误，还望海涵并指出。谢谢！

### 一、什么是远程过程调用 ###

#### 1.1、一体化调用 ####

当使用单机服务时，所有的接口、服务都集中在一台服务器上，此时进行的是单体的函数调用(只通过操作系统自身的堆、栈、代码段等)，接口之间的调用不需要通过网络来进行。

#### 1.2、远程过程调用 ####

单体服务被拆分为分布式服务时，接口之间的调用会存在跨机器的调用，此时需要借助网络来进行服务通信，然后进行远程调用。

**远程过程调用的形式有很多，最基础的是通过网络传递调用方的参数或序列化后的对象（严格地说，网络传输信道中只存在二进制流）传递给服务提供方，服务提供方接收到参数后调用本地接口进行处理（或再次调用远程服务），然后将产生的结果序列化后通过网络响应给调用方，这就是最基本的远程调用过程了。**

当然了，真正的RPC框架（dubbo、thrift）肯定会有更加复杂的实现，譬如请求重试、请求降级、网络拥塞判断等等，这里只是简单地实现核心的操作，重点在于如何通过Socket“远程调用”，方便加深对RPC的理解。

### 二、用Java实现简单RPC ###

首先，定义一个接口和实现类：

*  UserService

public interface UserService { 
        User getUser(long id);
        User addUser(long id, String name, int age);
    }

*  UserServiceImpl

public class UserServiceImpl implements UserService { 
        @Override
        public User getUser(long id) { 
            // 获取User
            System.out.println("Server:getUser");
            return new User(id, "DtnamicRpcTest", 20);
        }
    
        @Override
        public User addUser(long id, String name, int age) { 
            // 添加User
            System.out.println("Server:addUser");
            return new User(id, name, age);
        }
    }

#### 2.1、业务侵入型Socket ####

*  Client

public class Client { 
        public static void main(String[] args) throws IOException { 
            try (
                    Socket socket = new Socket("127.0.0.1", 8888);
                    ByteArrayOutputStream bos = new ByteArrayOutputStream();
                    DataOutputStream dos = new DataOutputStream(bos);
            ) { 
                // 获取id为10的User
                dos.writeLong(10);
                // 通过socket远程调用server的生成user方法
                socket.getOutputStream().write(bos.toByteArray());
                socket.getOutputStream().flush();
                // 接受响应
                try (
                        DataInputStream dis = new DataInputStream(socket.getInputStream());
                ) { 
                    long id = dis.readLong();
                    int age = dis.readInt();
                    String name = dis.readUTF();
                    User user = new User(id, name, age);
                    System.out.println("通过RPC调用Server方法，获取到User：" + user);
                } catch (IOException e) { 
                    e.printStackTrace();
                }
            } catch (IOException e) { 
                e.printStackTrace();
            }
    
        }
    }

*  Server

public class Server { 
        public static void main(String[] args) throws IOException { 
            ServerSocket serverSocket = new ServerSocket(8888);
            while (true) { 
                Socket socket = serverSocket.accept();
                process(socket);
                socket.close();
            }
        }
    
        private static void process(Socket socket) throws IOException { 
            try (
                    DataOutputStream dos = new DataOutputStream(socket.getOutputStream());
                    DataInputStream dis = new DataInputStream(socket.getInputStream());
            ) { 
                long id = dis.readLong();
                // 调用server的生成user方法
                System.out.println("Server的服务被调用");
                UserService userService = new UserServiceImpl();
                User user = userService.getUser(id);
                // 通过该sokcet将user所需的id、name和age返回
                dos.writeLong(user.getId());
                dos.writeInt(user.getAge());
                dos.writeUTF(user.getName());
                // 写回
                dos.flush();
            } catch (IOException e) { 
                e.printStackTrace();
            }
        }
    }

#### 2.2、无业务侵入型Socket ####

*  Client

public class Client { 
        public static void main(String[] args) throws IOException { 
            // 使用stub封装底层通信细节
            Stub stub = new Stub();
            stub.getUser(10);
        }
    }

*  Server

public class Server { 
        public static void main(String[] args) throws IOException { 
            ServerSocket serverSocket = new ServerSocket(8888);
            while (true) { 
                Socket socket = serverSocket.accept();
                process(socket);
                socket.close();
            }
        }
    
        private static void process(Socket socket) throws IOException { 
            DataOutputStream dos = new DataOutputStream(socket.getOutputStream());
            DataInputStream dis = new DataInputStream(socket.getInputStream());
            long id = dis.readLong();
            // 调用server的生成user方法
            System.out.println("Server的服务被调用");
            UserService userService = new UserServiceImpl();
            User user = userService.getUser(id);
            // 通过该sokcet将user所需的id、name和age返回
            dos.writeLong(user.getId());
            dos.writeInt(user.getAge());
            dos.writeUTF(user.getName());
            // 写回
            dos.flush();
        }
    }

*  Stub

public class Stub { 
        public User getUser(long id) throws IOException { 
            Socket socket = new Socket("127.0.0.1", 8888);
            ByteArrayOutputStream bos = new ByteArrayOutputStream();
            DataOutputStream dos = new DataOutputStream(bos);
            // 获取特定id的User
            dos.writeLong(id);
            // 通过socket远程调用server的生成user方法
            socket.getOutputStream().write(bos.toByteArray());
            socket.getOutputStream().flush();
            // 接受响应
            DataInputStream dis = new DataInputStream(socket.getInputStream());
            int age = dis.readInt();
            String name = dis.readUTF();
            User user = new User(id, name, age);
            System.out.println("通过RPC调用Server方法，获取到User："+ user);
            dos.close();
            dis.close();
            socket.close();
            return user;
        }
    }

#### 2.3、动态代理无业务侵入型Socket ####

*  Client

public class Client { 
        public static void main(String[] args) throws IOException { 
            UserService userService = (UserService) Stub.getStub(UserService.class);
            User user1 = userService.getUser(10);
            System.out.println(user1);
            User user2 = userService.addUser(11, "ARong", 21);
            System.out.println(user2);
        }
    }

*  Server

public class Server { 
        public static void main(String[] args) throws Exception { 
            ServerSocket serverSocket = new ServerSocket(8888);
            while (true) { 
                Socket socket = serverSocket.accept();
                process(socket);
                socket.close();
            }
        }
    
        private static void process(Socket socket) throws Exception { 
            System.out.println("Server的服务被调用");
            // 使用try-with-resource的方式处理资源，无需手动close
            try (
                    InputStream in = socket.getInputStream();
                    ObjectInputStream ois = new ObjectInputStream(in);
                    ObjectOutputStream oos = new ObjectOutputStream(socket.getOutputStream());
            ) { 
                // 获取到client端传过来的接口名+方法名+参数类型列表+参数列表
                String interfaceName = ois.readUTF();
                String methodName = ois.readUTF();
                System.out.println("server:获取到的接口名称为：" + interfaceName + ",方法名为：" + methodName);
                Class[] types = (Class[]) ois.readObject();
                Object[] args = (Object[]) ois.readObject();
                // 根据注册表查到该接口对应的具体业务实现类,然后通过反射获取该实现类(这里忽略了)
                UserServiceImpl userService = new UserServiceImpl();
                Method method = userService.getClass().getMethod(methodName, types);
                User user = (User) method.invoke(userService, args);
                // 将user对象返回
                oos.writeObject(user);
                // 追加空对象
                oos.writeObject(null);
                oos.flush();
            } catch (IOException e) { 
                e.printStackTrace();
            }
        }
    }

*  Stub

public class Stub { 
    
    
        /* * @Author ARong * @Description 使用clazz获取指定代理对象 * @Date 2020/4/29 1:09 下午 * @Param [clazz] * @return java.lang.Object **/
        public static Object getStub(Class clazz) { 
            InvocationHandler h = new InvocationHandler() { 
                @Override
                public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { 
                    // 构建代理方法
                    try (
                            Socket socket = new Socket("127.0.0.1", 8888);
                            ByteArrayOutputStream bos = new ByteArrayOutputStream();
                            ObjectOutputStream oos = new ObjectOutputStream(bos);
                    ) { 
                        // 获取当前的接口名称
                        String interfaceName = clazz.getName();
                        // 获取需要调用的方法名和参数类型列表
                        String methodName = method.getName();
                        Class<?>[] parameterTypes = method.getParameterTypes();
                        // 将接口名+方法名+参数类型列表+参数列表写给server端
                        oos.writeUTF(interfaceName);
                        oos.writeUTF(methodName);
                        oos.writeObject(parameterTypes);
                        oos.writeObject(args);
                        // 追加空对象，防止读取时出现EOF异常
                        oos.writeObject(null);
                        socket.getOutputStream().write(bos.toByteArray());
                        socket.getOutputStream().flush();
                        // 获取到返回的user对象
                        try (ObjectInputStream ois = new ObjectInputStream(socket.getInputStream());) { 
                            User user = (User) ois.readObject();
                            return user;
                        }
                    } catch (IOException e) { 
                        e.printStackTrace();
                    }
                    return null;
                }
    
            };
            // 生成代理对象
            Object o = Proxy.newProxyInstance(clazz.getClassLoader(), new Class[]{ clazz}, h);
            return o;
        }
    }

#### 2.4、Hessian序列化协议 ####

*  TestHessian

public class TestHessian { 
        public static void main(String[] args) throws IOException { 
            hessian();
        }
    
        private static void hessian() throws IOException { 
            User user = new User(1, "HessianTest", 20);
            ByteArrayOutputStream bis1 = new ByteArrayOutputStream();
            // 使用JDK序列化
            ObjectOutputStream oos = new ObjectOutputStream(bis1);
            oos.writeObject(user);
            oos.flush();
            System.out.println("JDK序列化字节长度："+bis1.toByteArray().length);
            // 使用hessian序列化
            ByteArrayOutputStream bis2 = new ByteArrayOutputStream();
            Hessian2Output hop = new Hessian2Output(bis2);
            hop.writeObject(user);
            hop.flush();
            System.out.println("Hessian序列化字节长度："+bis2.toByteArray().length);
            Hessian2Input hit = new Hessian2Input();
            ByteArrayInputStream bis = new ByteArrayInputStream(bis2.toByteArray());
            hit.init(bis);
            System.out.println(hit.readObject());
         }
    }

#### 2.5、RMI远程方法调用协议 ####

https://juejin.im/post/5bdfa4c76fb9a049e062c6ea

#### 2.6、Dubbo服务治理框架 ####

https://dubbo.apache.org/zh-cn/index.html