手写RPC之基于TCP

原创

用户10198968

发布于 2023-01-02 17:10:50

4980

文章被收录于专栏：SpringBoot原创文章系列SpringBoot原创文章系列

手写RPC之基于TCP

0x01_解决服务在远端问题

代码结构：

│   │   ├── main
│   │   │   ├── java
│   │   │   │   └── com
│   │   │   │       └── bones
│   │   │   │           └── rpc
│   │   │   │               ├── IUserService.java
│   │   │   │               ├── User.java
│   │   │   │               ├── client
│   │   │   │               │   └── FgClient.java
│   │   │   │               └── server
│   │   │   │                   ├── FgServer.java
│   │   │   │                   └── UserServiceImpl.java

实体类User

package com.bones.rpc;

import java.io.Serializable;

public class User implements Serializable {

    private static final long serialVersionUID = -1189354900598453538L;
    private Integer id;
    private String name;

    public User(Integer id, String name) {
        this.id = id;
        this.name = name;
    }

    public Integer getId() {
        return id;
    }

    public void setId(Integer id) {
        this.id = id;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    @Override
    public String toString() {
        return "User{" +
                "id=" + id +
                ", name='" + name + '\'' +
                '}';
    }
}

接口IUserService

package com.bones.rpc;

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

接口的实现server/UserServiceImpl

package com.bones.rpc.server;

import com.bones.rpc.IUserService;
import com.bones.rpc.User;

public class UserServiceImpl implements IUserService {

    @Override
    public User findUserById(Integer id) {
        return new User(id,"Bones");
    }
}

服务端server/FgServer

package com.bones.rpc.server;


import com.bones.rpc.IUserService;
import com.bones.rpc.User;

import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;

public class FgServer {
    public static void main(String[] args) throws IOException {
        ServerSocket serverSocket=new ServerSocket(9999);
        while (true){
            Socket socket=serverSocket.accept();
            process(socket);
            socket.close();
        }
    }
    private static void process(Socket socket) throws IOException {
        InputStream inputStream= socket.getInputStream();
        OutputStream outputStream=socket.getOutputStream();
        DataInputStream dataInputStream = new DataInputStream(inputStream);
        DataOutputStream dataOutputStream = new DataOutputStream(outputStream);
        System.out.println("process");
        int id = dataInputStream.readInt();
        IUserService service=new UserServiceImpl();
        User user = service.findUserById(id);
        dataOutputStream.writeInt(user.getId());
        dataOutputStream.writeUTF(user.getName());
        dataOutputStream.flush();
    }
}

客户端client/FgClient

package com.bones.rpc.client;

import com.bones.rpc.User;

import java.io.ByteArrayOutputStream;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.net.Socket;

public class FgClient {
    public static void main(String[] args) throws IOException {
        Socket socket = new Socket("127.0.0.1", 9999);
        // socket.getInputStream()

        ByteArrayOutputStream outputStream=new ByteArrayOutputStream();
        DataOutputStream dataOutputStream = new DataOutputStream(outputStream);
        dataOutputStream.writeInt(12);
        socket.getOutputStream().write(outputStream.toByteArray());
        socket.getOutputStream().flush();
        //dataOutputStream
        DataInputStream dataInputStream = new DataInputStream(socket.getInputStream());
        int id = dataInputStream.readInt();

        String name = dataInputStream.readUTF();
        User user=new User(id, name);
        System.out.println(user.getName());
        System.out.println(user.toString());
        outputStream.close();
        dataOutputStream.close();
        socket.close();
    }
}

0x02_解决客户端简单调用——引入存根stub

项目结构：

│   │   ├── main
│   │   │   ├── java
│   │   │   │   └── com
│   │   │   │       └── bones
│   │   │   │           └── rpc
│   │   │   │               ├── IUserService.java
│   │   │   │               ├── User.java
│   │   │   │               ├── client
│   │   │   │               │   ├── FgClient.java
│   │   │   │               │   └── Stub.java
│   │   │   │               └── server
│   │   │   │                   ├── FgServer.java
│   │   │   │                   └── UserServiceImpl.java

存根Stub

package com.bones.rpc.client;

import com.bones.rpc.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 socket = new Socket("127.0.0.1", 9999);
        // socket.getInputStream()

        ByteArrayOutputStream outputStream=new ByteArrayOutputStream();
        DataOutputStream dataOutputStream = new DataOutputStream(outputStream);
        dataOutputStream.writeInt(id);
        socket.getOutputStream().write(outputStream.toByteArray());
        socket.getOutputStream().flush();
        //dataOutputStream.
        DataInputStream dataInputStream = new DataInputStream(socket.getInputStream());
        id = dataInputStream.readInt();

        String name = dataInputStream.readUTF();
        //User user=new User(id, name);
        outputStream.close();
        dataOutputStream.close();
        socket.close();

        return new User(id,name);
    }
}

客户端FgClient

package com.bones.rpc.client;

import com.bones.rpc.User;
public class FgClient {
    public static void main(String[] args) throws Exception {
        Stub stub = new Stub();
        User user = stub.findUserById(124);
        System.out.println("user = " + user);
    }
}

客户端存根Stub：存放服务端的地址消息，再将客户端的请求参数打包成网络消息，然后通过网络远程发送给服务方。

0x03_客户端动态代理隐藏网络细节

Stub

package com.bones.rpc.client;

import com.bones.rpc.IUserService;
import com.bones.rpc.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;

public class Stub {

    static IUserService getStub(){
        InvocationHandler handler = new InvocationHandler() {
            @Override
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                Socket socket = new Socket("127.0.0.1", 9999);
                // socket.getInputStream()

                ByteArrayOutputStream outputStream=new ByteArrayOutputStream();
                DataOutputStream dataOutputStream = new DataOutputStream(outputStream);
                dataOutputStream.writeInt(12);
                socket.getOutputStream().write(outputStream.toByteArray());
                socket.getOutputStream().flush();
                //dataOutputStream.
                DataInputStream dataInputStream = new DataInputStream(socket.getInputStream());
                int id = dataInputStream.readInt();

                String name = dataInputStream.readUTF();
                //User user=new User(id, name);
                outputStream.close();
                dataOutputStream.close();
                socket.close();

                return new User(id,name);
            }
        };

        //执行动态代理
        Object o = Proxy.newProxyInstance(IUserService.class.getClassLoader(), new Class[]{IUserService.class}, handler);

        return (IUserService) o;
    }

}

客户端调用

FgClient

package com.bones.rpc.client;

import com.bones.rpc.IUserService;
import com.bones.rpc.User;


public class FgClient {
    public static void main(String[] args) throws Exception {
        IUserService service = Stub.getStub();
        User user = service.findUserById(13);
        System.out.println(user);
    }
}

此时id是写死的，后面用反射就可以传入id了。

0x04_服务端反射提高灵活性

存根Stub

package com.bones.rpc.client;

import com.bones.rpc.IUserService;
import com.bones.rpc.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;

public class Stub {

    static IUserService getStub(){
        InvocationHandler handler = new InvocationHandler() {
            @Override
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                Socket socket = new Socket("127.0.0.1", 9999);
                // socket.getInputStream()

                ObjectOutputStream objectOutputStream = new ObjectOutputStream(socket.getOutputStream());
                String methodName= method.getName();//方法
                Class[] parametersTypes = method.getParameterTypes();//参数类型
                //格式1：方法名、2、方法参数类型  3、参数值
                objectOutputStream.writeUTF(methodName);
                objectOutputStream.writeObject(parametersTypes);
                objectOutputStream.writeObject(args);
                objectOutputStream.flush();

                //dataOutputStream.
                DataInputStream dataInputStream = new DataInputStream(socket.getInputStream());
                int id = dataInputStream.readInt();

                String name = dataInputStream.readUTF();
                objectOutputStream.close();
                socket.close();

                return new User(id, name);
            }
        };

        //执行动态代理
        Object o = Proxy.newProxyInstance(IUserService.class.getClassLoader(), new Class[]{IUserService.class}, handler);

        return (IUserService) o;
    }
}

服务类FgServer

package com.bones.rpc.server;


import com.bones.rpc.IUserService;
import com.bones.rpc.User;

import java.io.*;
import java.lang.reflect.Method;
import java.net.ServerSocket;
import java.net.Socket;

public class FgServer {
    public static void main(String[] args) throws Exception {
        ServerSocket serverSocket=new ServerSocket(9999);
        while (true){
            Socket socket=serverSocket.accept();
            process(socket);
            socket.close();
        }
    }
    private static void process(Socket socket) throws Exception {
        InputStream inputStream= socket.getInputStream();
        OutputStream outputStream=socket.getOutputStream();
        ObjectInputStream dataInputStream = new ObjectInputStream(inputStream);
        DataOutputStream dataOutputStream = new DataOutputStream(outputStream);
        System.out.println("process");

        //格式1：方法名、2、方法参数类型  3、参数值
        String methodName = dataInputStream.readUTF();//方法名
        Class[] parametersTypes = (Class[]) dataInputStream.readObject();//方法参数类型
        Object[] args=(Object[]) dataInputStream.readObject();//参数值

        IUserService service=new UserServiceImpl();
        //反射
        Method method = service.getClass().getMethod(methodName, parametersTypes);
        User user = (User) method.invoke(service, args);
        dataOutputStream.writeInt(user.getId());
        dataOutputStream.writeUTF(user.getName());
        dataOutputStream.flush();
    }
}

测试，先运行服务端，再运行客户端：

package com.bones.rpc.client;

import com.bones.rpc.IUserService;
import com.bones.rpc.User;


public class FgClient {
    public static void main(String[] args) throws Exception {
        IUserService service = Stub.getStub();
        User user = service.findUserById(137);
        System.out.println(user);
    }
}

0x05_灵活的传递参数

存根Stub

package com.bones.rpc.client;

import com.bones.rpc.IUserService;
import com.bones.rpc.User;

import java.io.*;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.lang.reflect.Proxy;
import java.net.Socket;

public class Stub {

    static IUserService getStub(Class clazz){
        InvocationHandler handler = new InvocationHandler() {
            @Override
            public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
                Socket socket = new Socket("127.0.0.1", 9999);
                // socket.getInputStream()

                ObjectOutputStream objectOutputStream = new ObjectOutputStream(socket.getOutputStream());
                String methodName= method.getName();//方法
                Class[] parametersTypes = method.getParameterTypes();//参数类型
                //格式（0：类名，1：方法名、2、方法参数类型  3、参数值）
                objectOutputStream.writeUTF(clazz.getName());
                objectOutputStream.writeUTF(methodName);
                objectOutputStream.writeObject(parametersTypes);
                objectOutputStream.writeObject(args);
                objectOutputStream.flush();

                //dataOutputStream.
                ObjectInputStream dataInputStream = new ObjectInputStream(socket.getInputStream());

                Object o = dataInputStream.readObject();


                objectOutputStream.close();
                socket.close();

                return o;
            }
        };

        //执行动态代理
        Object o = Proxy.newProxyInstance(IUserService.class.getClassLoader(), new Class[]{IUserService.class}, handler);

        return (IUserService) o;
    }

}

FgServer

package com.bones.rpc.server;


import com.bones.rpc.IUserService;
import com.bones.rpc.User;

import java.io.*;
import java.lang.reflect.Method;
import java.net.ServerSocket;
import java.net.Socket;

public class FgServer {
    public static void main(String[] args) throws Exception {
        ServerSocket serverSocket=new ServerSocket(9999);
        while (true){
            Socket socket=serverSocket.accept();
            process(socket);
            socket.close();
        }
    }
    private static void process(Socket socket) throws Exception {
        InputStream inputStream= socket.getInputStream();
        OutputStream outputStream=socket.getOutputStream();
        ObjectInputStream dataInputStream = new ObjectInputStream(inputStream);
        ObjectOutputStream dataOutputStream = new ObjectOutputStream(outputStream);
        System.out.println("process");

        //格式（0：类名，1：方法名、2、方法参数类型  3、参数值）
        String clazzName = dataInputStream.readUTF();
        String methodName = dataInputStream.readUTF();//方法名
        Class[] parametersTypes = (Class[]) dataInputStream.readObject();//方法参数类型
        Object[] args=(Object[]) dataInputStream.readObject();//参数值


        //反射
        //从服务的注册表中寻找（注册中心）
        Class clazz = UserServiceImpl.class;
        Method method = clazz.getMethod(methodName, parametersTypes);
        User user = (User) method.invoke(clazz.newInstance(), args);
        dataOutputStream.writeObject(user );
        dataOutputStream.flush();
    }
}

FgClient

package com.bones.rpc.client;

import com.bones.rpc.IUserService;
import com.bones.rpc.User;


public class FgClient {
    public static void main(String[] args) throws Exception {
        IUserService service = Stub.getStub(IUserService.class);
        User user = service.findUserById(98);
        System.out.println(user);
    }
}

测试结果：

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

https