30行代码透彻解析RPC

淡淡的烟草味﹌ 2024-04-07 14:11 76阅读 0赞

从单机到分布式 -> 分布式通信 ->最基本：二进制数据传输 TCP/IP

## 1、基本代码 ##

package com.mashibing.rpc.common;
    
    import java.io.Serializable;
    
    public class User implements Serializable {
        
        private static final long serialVersionUID = 1L;
    
        private Integer id;
        private String name;
    
        public User(Integer id, String name) {
        
            this.id = id;
            this.name = name;
        }
    
        public Integer getId() {
        
            return id;
        }
    
        public String getName() {
        
            return name;
        }
    
        public void setId(Integer id) {
        
            this.id = id;
        }
    
        public void setName(String name) {
        
            this.name = name;
        }
    
        @Override
        public String toString() {
        
            return "User{" +
                    "id=" + id +
                    ", name='" + name + '\'' +
                    '}';
        }
    }

package com.mashibing.rpc.common;
    
    public interface IUserService {
        
        public User findUserById(Integer id);
    }

package com.mashibing.rpc01;
    
    
    import com.mashibing.rpc.common.IUserService;
    import com.mashibing.rpc.common.User;
    
    public class UserServiceImpl implements IUserService {
        
        @Override
        public User findUserById(Integer id) {
        
            return new User(id, "Alice");
        }
    }

package com.mashibing.rpc01;
    
    import com.mashibing.rpc.common.User;
    
    import java.io.ByteArrayOutputStream;
    import java.io.DataInputStream;
    import java.io.DataOutputStream;
    import java.net.Socket;
    
    public class Client {
        
        public static void main(String[] args) throws Exception {
        
            Socket s = new Socket("127.0.0.1", 8888);
            ByteArrayOutputStream baos = new ByteArrayOutputStream();
            DataOutputStream dos = new DataOutputStream(baos);
            dos.writeInt(123);
    
            s.getOutputStream().write(baos.toByteArray());
            s.getOutputStream().flush();
    
            DataInputStream dis = new DataInputStream(s.getInputStream());
            int id = dis.readInt();
            String name = dis.readUTF();
            User user = new User(id, name);
    
            System.out.println(user);
    
            dos.close();
            s.close();
        }
    }

## 2、增加stub，提供网络接口 ##

package com.mashibing.rpc02;
    
    import com.mashibing.rpc.common.User;
    
    import java.io.ByteArrayOutputStream;
    import java.io.DataInputStream;
    import java.io.DataOutputStream;
    import java.net.Socket;
    
    public class Stub {
        
        public User findUserById(Integer id) throws Exception {
        
            Socket s = new Socket("127.0.0.1", 8888);
            ByteArrayOutputStream baos = new ByteArrayOutputStream();
            DataOutputStream dos = new DataOutputStream(baos);
            dos.writeInt(123);
    
            s.getOutputStream().write(baos.toByteArray());
            s.getOutputStream().flush();
    
            DataInputStream dis = new DataInputStream(s.getInputStream());
            int receivedId = dis.readInt();
            String name = dis.readUTF();
            User user = new User(id, name);
    
            dos.close();
            s.close();
            return user;
        }
    }

## 3、动态代理 ##

package com.mashibing.rpc03;
    
    import com.mashibing.rpc.common.IUserService;
    import com.mashibing.rpc.common.User;
    
    import java.io.ByteArrayOutputStream;
    import java.io.DataInputStream;
    import java.io.DataOutputStream;
    import java.lang.reflect.InvocationHandler;
    import java.lang.reflect.Method;
    import java.lang.reflect.Proxy;
    import java.net.Socket;
    
    /**
     * 而且Client的调用显得不是很合理（Stub里只有findById的代码），如果有个findByName的新方法，那么就又得重新改进
     * 下面这种写法解决了方法增加的问题
     */
    
    public class Stub {
        
        public static IUserService getStub() {
        
            InvocationHandler h = new InvocationHandler() {
        
                @Override
                public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        
                    Socket s = new Socket("127.0.0.1", 8888);
    
                    ByteArrayOutputStream baos = new ByteArrayOutputStream();
                    DataOutputStream dos = new DataOutputStream(baos);
                    dos.writeInt(123);
    
                    s.getOutputStream().write(baos.toByteArray());
                    s.getOutputStream().flush();
    
                    DataInputStream dis = new DataInputStream(s.getInputStream());
                    int id = dis.readInt();
                    String name = dis.readUTF();
                    User user = new User(id, name);
    
                    dos.close();
                    s.close();
                    return user;
                }
            };
            Object o = Proxy.newProxyInstance(IUserService.class.getClassLoader(), new Class[] {
        IUserService.class}, h);
            return (IUserService)o;
        }
    }

## 4、方法代理再升级 ##

package com.mashibing.rpc04;
    
    import com.mashibing.rpc.common.IUserService;
    import com.mashibing.rpc.common.User;
    
    import java.io.ByteArrayOutputStream;
    import java.io.DataInputStream;
    import java.io.DataOutputStream;
    import java.io.ObjectOutputStream;
    import java.lang.reflect.InvocationHandler;
    import java.lang.reflect.Method;
    import java.lang.reflect.Proxy;
    import java.net.Socket;
    
    /**
     * 但是这里仅仅实现了findByUserId的方法代理，如果要实现其他方法的代理该怎么做呢？
     * 这里就要从协议层做出改进
     *
     * 服务器端也要做出对应处理
     */
    
    public class Stub {
        
        public static IUserService getStub() {
        
            InvocationHandler h = new InvocationHandler() {
        
                @Override
                public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        
                    Socket s = new Socket("127.0.0.1", 8888);
    
                    ObjectOutputStream oos = new ObjectOutputStream(s.getOutputStream());
    
                    String methodName = method.getName();
                    Class[] parametersTypes = method.getParameterTypes();
                    oos.writeUTF(methodName);
                    oos.writeObject(parametersTypes);
                    oos.writeObject(args);
                    oos.flush();
    
    
                    DataInputStream dis = new DataInputStream(s.getInputStream());
                    int id = dis.readInt();
                    String name = dis.readUTF();
                    User user = new User(id, name);
    
                    oos.close();
                    s.close();
                    return user;
                }
            };
            Object o = Proxy.newProxyInstance(IUserService.class.getClassLoader(), new Class[] {
        IUserService.class}, h);
            return (IUserService)o;
        }
    
    
    
    
    }

package com.mashibing.rpc04;
    
    import com.mashibing.rpc.common.IUserService;
    import com.mashibing.rpc.common.User;
    
    import java.io.*;
    import java.lang.reflect.Method;
    import java.net.ServerSocket;
    import java.net.Socket;
    
    public class Server {
        
        private static boolean running = true;
        public static void main(String[] args) throws Exception {
        
            ServerSocket ss = new ServerSocket(8888);
            while (running) {
        
                Socket s = ss.accept();
                process(s);
                s.close();
            }
            ss.close();
        }
    
        private static void process(Socket s) throws Exception {
        
            InputStream in = s.getInputStream();
            OutputStream out = s.getOutputStream();
            ObjectInputStream oos = new ObjectInputStream(in);
            DataOutputStream dos = new DataOutputStream(out);
    
            String methodName = oos.readUTF();
            Class[] parameterTypes = (Class[])oos.readObject();
            Object[] args = (Object[])oos.readObject();
    
            IUserService service = new UserServiceImpl();
            Method method = service.getClass().getMethod(methodName, parameterTypes);
            User user = (User)method.invoke(service, args);
    
    
    
            dos.writeInt(user.getId());
            dos.writeUTF(user.getName());
            dos.flush();
        }
    }

## 5、Object 封装 ##

返回值用Object封装，支持任意类型