[网络协议]简单 RPC 实现(一)

简单 RPC 实现(一)

项目结构

[root@localhost ppr-parent]# tree -L 1
├── pom.xml
├── ppr-consumer  # 消费者
├── ppr-provider  # 服务提供者
├── ppr-sdk       # rpc sdk
└── ppr-service   # 服务接口定义

ppr-consumer 依赖 ppr-sdk,ppr-service
ppr-provider 依赖 ppr-sdk,ppr-service

展开

├── pom.xml
├── ppr-consumer
│   ├── pom.xml
│   └── src
│       ├── main
│       │   └── java
│       │       └── com.ddup.consumer
│       │                   └── service
│       │                       └── HelloServicePpr.java

├── ppr-provider
│   ├── pom.xml
│   └── src
│       ├── main
│       │   └── java
│       │       └── com.ddup.provider
│       │                   ├── server
│       │                   │   └── Server.java
│       │                   └── service
│       │                       └── HelloServiceImpl.java

├── ppr-sdk
│   ├── pom.xml
│   └── src
│       ├── main
│       │   └── java
│       │       └── com.ddup. sdk
│       │                   ├── exception
│       │                   │   └── RpcException.java
│       │                   ├── handler
│       │                   │   ├── NetClient.java
│       │                   │   └── PprProxyInvocationHandler.java # 客户端代理
│       │                   ├── PprRequest.java
│       │                   └── server
│       │                       └── ProviderServer.java

└── ppr-service
    ├── pom.xml
    └── src
        ├── main
        │   └── java
        │       └── com.ddup. service
        │                   └── HelloService.java

ppr-service 定义接口

public interface HelloService {
    String sayHi(String name);
}

ppr-consumer 消费者

• 定义 PprRequest 【一定要实现序列化，对象内部的属性也必须是实现序列化】

• 创建代理类

• 使用socket 进行数据传输

PprRequest

@Data
public class PprRequest implements Serializable {

    private Object[] args;
    private Class<?> clz;
    // Method 使用这个属性不行， Method 没有实现序列化
    //private Method method;
    private String method;
    private Class<?>[] parameterTypes;

}

请求代理类 PprProxyInvocationHandler

所有的接口都通过代理类调用 socket 通信

public class PprProxyInvocationHandler<T> implements InvocationHandler {

    private Class<T> service;
    private NetClient client = new NetClient();

    public PprProxyInvocationHandler(Class<T> service) {
        this.service = service;
    }

    public T getProxy() {
        //如果是接口具体的实现类，第二个参数用 service.getInterfaces()
        //return (T) Proxy.newProxyInstance(service.getClassLoader(), service.getInterfaces(), this);
        //如果只是接口，第二个参数使用 new Class[]{service}
        return (T) Proxy.newProxyInstance(service.getClassLoader(), new Class[]{service}, this);
    }

    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        PprRequest request = new PprRequest();
        request.setArgs(args);
        request.setMethod(method.getName());
        request.setParameterTypes(method.getParameterTypes());
        request.setClz(service);
        return client.execute(request);
    }
}

Rpc 通信 NetClient

public class NetClient {

    public Object execute(PprRequest request) {
        ObjectInputStream ois = null;
        ObjectOutputStream oos = null;
        try {
            String host = "localhost";
            int port = 8888;
            Socket socket = new Socket(host, port);
            oos = new ObjectOutputStream(socket.getOutputStream());
            oos.writeObject(request);
            oos.flush();
            ois = new ObjectInputStream(socket.getInputStream());
            Object result = ois.readObject();
            return result;
        } catch (Exception e) {
            try {
                if (ois != null) {
                    ois.close();
                }
                if (oos != null) {
                    oos.close();
                }
            } catch (IOException ex) {
                ex.printStackTrace();
            }
        }
        throw new RpcException("ppr调用异常");
    }
}

客户端调用:

public class HelloServicePpr {

    public void helloTest() {
        String name = "小明";
        PprProxyInvocationHandler invocationHandler = new PprProxyInvocationHandler(HelloService.class);
        HelloService proxy = (HelloService) invocationHandler.getProxy();
        String s = proxy.sayHi(name);
        System.out.println(s);
    }

    public static void main(String[] args) {
        new HelloServicePpr().helloTest();
    }
}

服务提供者 ppr-provider

• socket 反序列化 生成 PprRequest
• 根据 PprRequest 中 className 找到具体实现类
• 使用反射执行方法
• 序列化结果返回给消费者

服务端通信模板

public abstract class ProviderServer {

    public void start() {
        ObjectInputStream ois = null;
        ObjectOutputStream oos = null;
        try {
            int port = 8888;
            ServerSocket serverSocket = new ServerSocket(port);
            System.out.println("ppr server start at 8888...");
            while (true) {
                try {
                    Socket socket = serverSocket.accept();
                    //反序列化
                    ois = new ObjectInputStream(socket.getInputStream());
                    PprRequest request = (PprRequest) ois.readObject();
                    //反射调用
                    Object result = invoke(request);
                    oos = new ObjectOutputStream(socket.getOutputStream());
                    oos.writeObject(result);
                    oos.flush();
                } catch (Exception e) {
                    try {
                        if (ois != null) {
                            ois.close();
                        }
                        if (oos != null) {
                            oos.close();
                        }
                    } catch (IOException ex) {
                        ex.printStackTrace();
                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    protected Object invoke(PprRequest request) {
        try {
            Method method = request.getClz().getMethod(request.getMethod(), request.getParameterTypes());
            Object o = getProvider(request);
            if (o != null) {
                return method.invoke(o, request.getArgs());
            } else {
                System.out.println("bean 不存在");
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 获取实现类
     *
     * @param request
     * @return
     */
    protected abstract Object getProvider(PprRequest request);
}

1. 在 ppr-provider 中 启动服务:

public class Server extends ProviderServer {

    private ConcurrentHashMap<String, Object> factory = new ConcurrentHashMap<>();

    {
        factory.put("com.ddup.service.HelloService", new HelloServiceImpl());
    }

    public static void main(String[] args) throws IOException {
        new Server().start();
        System.in.read();
    }

    @Override
    protected Object getProvider(PprRequest request) {
        return factory.get(request.getClz().getName());
    }
}

2.HelloService 实现业务

public class HelloServiceImpl implements HelloService {

    @Override
    public String sayHi(String name) {
        return "echo from server, hi: " + name;
    }
}

启动 ppr-provider 服务

ppr server start at 8888...

ppr-consumer 调用

echo from server, hi: 小明

good luck!