RPC基础理解-有人思眼前，有人思来年

RPC简述

RPC是什么？

RPC (Remote Procedure Call)远程过程调用，这只是一个统称，重点在于方法调用(不支持对象的概念)，具体实现甚至可以用RMI, RestFul等去实现，但一般不用，因为RMI不能跨语言，而RestFul效率太低。多用于服务器集群间的通信，因此常使用更加高效短小精悍的传输模式以提高效率。

从单机到分布式-->分布式通信 ,RPC就是分布式通信的一种方式

新建两个两个实现了序列化的基类User，Product：

User.java

import java.io.Serializable;
public class User implements Serializable {
    private static final long serialVersionUID = 1L;
    
    private Integer id;
    private Stringname
    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 +  '\'+
            '}'
   }
}

IUserService.java

public interface IUserService {
    public User  findUserById(Integer id);
}

Product.java

import java.io.Serializable;
public class Product implements Serializable{
    private static final long serialVersionUID = 1L;
    
    private Integer id;
    
    private String name;
    
    public Product(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(){}
}

IProductService.java

public interface IProductService{
    public Product findProductById(Integer id);
}

IUserServiceImpl和IProductServiceImpl.java存放在Server端

IUserServiceImpl

public class IUserServiceImpl implements IUserService {
    @Override
    public User findUserById(Integer id) { 
        return new User(id, "name");
    }
}

IProductServiceImpl.java

public class IProductServiceImpl implements IProductService {
    @Override
    public User findProductById(Integer id) { 
        return new Product(id, "name");
    }
}

目标：

通过简单的七次迭代大概了解RPC底层实现流程

RPC实现方式：

通过动态代理等方式实现多类型多方法的服务器之间的远程调用（ps:了解给定api接口的情况下，服务器之间的通信底层是怎么运行的）

rpc-v0.1

整体流程为Client 发送123到Server，Server端accept到数据进行处理，根据123查询数据库返回对象属性给客户端，客户端再根据属性重新new成对象

弊端：传输过程和业务代码混合，且高度耦合，不容易修改

client.java

public class Client{
    public static void main(String[] args) throws Exception{
        Socket s = new Socket( host:"127.0..1", port: 8888);
        //网络上只能传输二进制，所以要创建一个bytearray 
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        DataOutputStream dos = new DataOutputStream(baos):
        dos,writeInt( v: 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.printIn(user);
        dos,close();
        s.close();
    }
}

Server.java

public class Server{
    private static boolean running = true;
    public static void main(String[] args) throws Exception {
        ServerSocket ss = new ServerSocket( port: 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();
        DataInputStream dis = new DataInputStream(in);
        DataOutputStream dos = new DataOutputStream(out);
        
        int id = dis.readInt();
        IUserService service = new UserServiceImpl();
        User user = service.findUserById(id);
        dos.writeInt(user.getId());
        dos,writeUTF(user.getName());
        dos.flush();
    }
}

rpc-v0.2

拆分客户端网络连接部分新建Stub类

Client.java

 public class Client{
     public static void main(String[] args) throws Exception{
         Stub stub = new Stub();
         System.out.println(stub.findUserById(123));
     }
 }

Stub.java

采用代理模式将网络链接部分写在一个新类里

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(id);
        
        s.getOutputstream().write(baos.toByteArray());
        s.getoutputstream().flush();
        DataInputstream dis = new DataInputstream(s.getInputstream0));
        int receivedId = dis.readIntO;
        String name = dis.readuTFO;
        User user = new User(receivedId, name);
        
        dos.c1ose();
        s.close();
        return user;
    }
}

Server.java

public class Server{
    private static boolean running = true;
    public static void main(String[] args) throws Exception {
        ServerSocket ss = new ServerSocket( port: 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();
        DataInputStream dis = new DataInputStream(in);
        DataOutputStream dos = new DataOutputStream(out);
        
        int id = dis.readInt();
        IUserService service = new UserServiceImpl();
        User user = service.findUserById(id);
        dos.writeInt(user.getId());
        dos,writeUTF(user.getName());
        dos.flush();
    }
}

rpc-v0.3

修改rpc02中只代理了一个类的问题，设计到代理模式里面的动态代理，只查看Client的写法，不考虑具体实现。(ps:动态代理是实现RPC的核心关键之一)

Client.java


public class Client {
    public static void main(String[] args) throws Exception{
        //getStub()返回的是一个代理类
        IUserService service = Stub.getStub();
        //service.findUserById调用的是Stub.getstub.InvocationHandTer().invoke(o, findUserById(), 123)
        System.out.printn(service.findUserById(123));
    }
}

Stub.java

public Class Stub{
    public static IUserService getstub() {
            InvocationHandler h = new InvocationHandTer(){
                @override
                public object invoke(object proxy， Method method, object[] args) throws
        Throwable{
                    socket s = new socket("127.0.0.1"，8888);
                    ByteArrayOutputstream baos = new ByteArrayoutputstream0);
                    DataOutputstream dos = new DataOutputstream(baos);
                    dos.writeInt(args[0]);
                    
                    s.getoutputstream().write(baostoByteArray());
                    s.getoutputstream().flush();
                                              
                    DataInputstream dis = new DataInputstream(s.getInputstream());
                    int id = disreadInt();
                    String name = dis.readUTF();
                    User user = new User(id, name);
                                              
                    dos.close();
                    s.close();
                    return user;

                }
            };
            Object o= Proxy.newProxyInstance(IUserService.class.getclassLoader()，newClass[]{IUserService.class}，h);
            //为方便处理，这里输出参数进行调试
            System.out.println(o.getClass().getName());
            System.out.println(o.getClass().getInterface());
            return (IUserService)o;
        }
}

Server.java

public class Server{
    private static boolean running = true;
    public static void main(String[] args) throws Exception {
        ServerSocket ss = new ServerSocket( port: 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();
        DataInputStream dis = new DataInputStream(in);
        DataOutputStream dos = new DataOutputStream(out);
        
        int id = dis.readInt();
        IUserService service = new UserServiceImpl();
        User user = service.findUserById(id);
        dos.writeInt(user.getId());
        dos,writeUTF(user.getName());
        dos.flush();
    }
}

rpc-v0.4

修改Stub()只能通过写死的方式实现功能, 通过代理远程调用的方式调用服务器中的方法,同理Server端要新增处理操作

Client.java


public class Client {
    public static void main(String[] args) throws Exception{
        IUserService service = Stub.getStub();
        //findUserById()会调用Stub中的invoke方法
        System.out.printn(service.findUserById(123));
    }
}

Stub.java

public Class Stub{
    public static IUserService getstub() {
            InvocationHandler h = new InvocationHandTer(){
                @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[] parameterTypes = method.getParameterTypes();
                    
                    oos.writeUTF(methodName);
                    oos.writeObject(parameterTypes);
                    oos.writeObject(args);
                    oos.flush;    
                                 
                    DataInputstream dis = new DataInputstream(s.getInputstream());
                    int id = disreadInt();
                    String name = dis.readUTF();
                    User user = new User(id, name);
                                              
                    dos.close();
                    s.close();
                    return user;

                }
            };
            Object o= Proxy.newProxyInstance(IUserService.class.getclassLoader()，newClass[]{IUserService.class}，h);
            //为方便处理，这里输出参数进行调试
            System.out.println(o.getClass().getName());
            System.out.println(o.getClass().getInterface());
            return (IUserService)o;
        }
}

Server.java

public class Server{
    private static boolean running = true;
    public static void main(String[] args) throws Exception {
        ServerSocket ss = new ServerSocket( port: 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();
        DataInputStream ois = new ObjectInputStream(in);
        DataOutputStream dos = new DataOutputStream(out);
        
        String methodName = ois.readUTF();
        Class[] parameterTypes = (Class[])ois.readObject();
        Object[] args = (0bject[])ois.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();
    }
}

rpc-v0.5

返回值用Object封装，支持任意类型修改Stub 和Server代码

Client.java

public class Client {
    public static void main(String[] args) throws Exception{
        IUserService service = Stub.getStub();
        //findUserById()会调用Stub中的invoke方法
        System.out.printn(service.findUserById(123));
    }
}

Stub.java

public Class Stub{
    public static IUserService getstub() {
            InvocationHandler h = new InvocationHandTer(){
                @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[] parameterTypes = method.getParameterTypes();
                    oos.writeUTF(methodName);
                    oos.writeObject(parameterTypes);
                    oos.writeObject(args);
                    oos.flush;    
                    
                    ObjectInputStream ois = new ObjectInputStream(s.getInputstream());
                    User user = (User)ois.readObject();
                                              
                    oos.close();
                    s.close();
                    return user;
                }
            };
            Object o= Proxy.newProxyInstance(IUserService.class.getclassLoader()，newClass[]{IUserService.class}，h);
            return (IUserService)o;
        }
}

Server.java

public class Server{
    private static boolean running = true;
    public static void main(String[] args) throws Exception {
        ServerSocket ss = new ServerSocket( port: 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 ois = new ObjectInputStream(in);
        //DataOutputStream dos = new DataOutputStream(out);
        
        String methodName = ois.readUTF();
        Class[] parameterTypes = (Class[])ois.readObject();
        Object[] args = (0bject[])ois.readObject();
        
        IUserService service = new UserServiceImpl();
        Method method = service,getClass().getMethod(methodName,parameterTypes);
        User user = (User)method.invoke(service，args);
        
        ObjectOutputStream oos = new ObjectOutputStream(out);
        oos.writeObject(user);
        oos.flush();
        
    }
}

rpc-v0.6

更改只能生成一个类型（IUserService）的代理，帮助生成更多类型的代理
修改Client,Server,Stub (ps: 在设计系统时，首先从Client端进行考虑)

Client.java

public class Client {
    public static void main(String[] args) throws Exception{
        IUserService service = (IUserService)Stub.getStub(IUserService.class);
        //findUserById()会调用Stub中的invoke方法
        System.out.printn(service.findUserById(123));
    }
}

Stub.java

public Class Stub{
    //change IUservice to Object
    public static Object getstub(Class clazz) {
            InvocationHandler h = new InvocationHandTer(){
                @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 clazzName = clazz.getName();
                    //获取方法名
                    String methodName = method.getName();
                    //获取方法参数，用于方法重载
                    Class[] parameterTypes = method.getParameterTypes();
                    
                    //first read first read
                    oos.writeUTF(clazzName);
                    oos.writeUTF(methodName);
                    oos.writeObject(parameterTypes);
                    oos.writeObject(args);
                    oos.flush;    
                    
                    ObjectInputStream ois = new ObjectInputStream(s.getInputstream());
                    User user = (User)ois.readObject();
                                              
                    oos.close();
                    s.close();
                    return user;
                }
            };
            Object o= Proxy.newProxyInstance(IUserService.class.getclassLoader()，newClass[]{IUserService.class}，h);
            return (IUserService)o;
        }
}

Server.java

public class Server{
    private static boolean running = true;
    public static void main(String[] args) throws Exception {
        ServerSocket ss = new ServerSocket( port: 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();
        DataInputStream ois = new ObjectInputStream(in);
        
        //first read first read
        String clazzName = ois.readUTF();
        String methodName = ois.readUTF();
        Class[] parameterTypes = (Class[])ois.readObject();
        Object[] args = (0bject[])ois.readObject();
        
        Class clazz = null;
        clazz = UserServiceImpl.class;
  
        Method method = clazz.getMethod(methodName，parameterTypes);
        Object o= (0bject)method.invoke(clazz.newInstance()，args);
        
        ObjectOutputStream oos = new ObjectOutputStream(out);
        oos.writeObject(o);
        oos.flush();
    }
}

rpc-v0.7

测试0.6版本是否生效，Server,Stub 未作修改

Client.java

public class Client {
    public static void main(String[] args) throws Exception{
        IUserService service = (IProductService)Stub.getStub(IProductService.class);
        //findUserById()会调用Stub中的invoke方法
        System.out.printn(service.findProductById(23));
    }
}

rpc-v0.8--v0.9

RPC序列化框架

java.io.Serializable(之前的代码都使用的Serializable )
Hessian
google protobuf
facebook Thrift
kyro
fst
json序列化框架
Jackson
- google Gson
- Ali FastIson
xmlrpc(xstream)
.........

测试Hessian和Serializable

public class HelloHessian{
    public static void main(String[] args) throws Exception{
        User u = new User( id: 1, name: "zhangsan");  
        System.out.printIn("hessin" + serialize(u).length);
        System.out.printIn("jdk" + jdkSerialize(u).length);
        byte[] bytes = serialize(u);
        System.out.printIn(bytes.length);
        User u1 = (User)deserialize(bytes);
        System.out.println(u1);
    }
    
    public static byte[] serialize(Object o) throws Exception{
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        Hessian20utput output = new Hessian20utput(baos);
        
        output.write0bject(o);
        output,flush();
        byte[] bytes = baos.toByteArray();
        
        baos.close();
        output.close();
        
        return bytes;
    }
    
    public static Object deserialize(byte[] bytes) throws Exception{
        ByteArrayInputStream bais = new ByteArrayInputStream();
        Hessian2Input intput = new Hessian2Input(bais);
        
        Object o = intput.readObject();
        bais.close();
        intput.close();
        return bytes;
    }
    
    public static byte[] jdkSerialize(Object o) throws Exception{
        ByteArrayOutputStream baos = new ByteArrayOutputStream();
        ObjectOutputStream output = new ObjectOutputStream(baos);

        output.writeObject(o);
        output.flush();
        byte[] bytes = baos.toByteArray();

        baos.close();
        output.close();
        return bytes;
    }
    public static Object jdkSerialize(byte[] bytes) throws Exception{
        ByteArrayInputStream bais = new ByteArrayInputStream();
        ObjectInputStream input = new ObjectInputStream(bais);

        Object o = intput.readObject();
        bais.close();
        input.close(); 

        return o;
    }

}

课程来源：https://www.bilibili.com/video/BV1r3411Y7fd , p108-118

RPC基础理解

RPC简述

RPC是什么？

新建两个两个实现了序列化的基类User，Product：

IUserServiceImpl

目标：

RPC实现方式：

rpc-v0.1

rpc-v0.2

rpc-v0.3

rpc-v0.4

rpc-v0.5

rpc-v0.6

rpc-v0.7

rpc-v0.8--v0.9

RPC序列化框架

测试Hessian和Serializable

评论区