第五章手写RPC框架

前请提示
代码演示
测试

前请提示

  我们在项目中微服务之间是可以相互调用的，在调用过程中我们只需要将一个微服务名+接口路径+接口请求对象给到我们调用方，那么调用方便可以发起调用。那它这么一个过程是怎样实现的呢？<br />      其实根据我们学到的zk知识其实就很容易联想到应该是这么一个调用过程。首先我们服务A给服务B提供服务，服务A一般是集群部署，然后将服务A所提供的所有接口服务暴露在zk，然后服务B去找zk中是否有自己所需的服务，存在呢可能就会通过nginx负载均衡到一台服务器A，然后服务器A会去分发找到对应的请求，并将结果返回给服务B。<br />      我这里就手写和比较简单的BIO的请求，而且这里面还没写服务暴露发现，也没用到任何协议，也没有去更好的序列化传输机制，也没有使用更好的性能的通信框架NIO或者netty。仅仅是作为自己的笔记而已。<br />![屏幕快照 2020-03-29 下午1.09.41.png](https://cdn.nlark.com/yuque/0/2020/png/771792/1585459865539-f410aa12-74e1-4086-8510-4256e9b48515.png#crop=0&crop=0&crop=1&crop=1&height=261&id=rUiJv&name=%E5%B1%8F%E5%B9%95%E5%BF%AB%E7%85%A7%202020-03-29%20%E4%B8%8B%E5%8D%881.09.41.png&originHeight=261&originWidth=769&originalType=binary&ratio=1&rotation=0&showTitle=false&size=43537&status=done&style=none&title=&width=769)

代码演示

自定义注解仿照spring

@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.METHOD)
public @interface GetMappingByHmm {
    String value();
}
@Retention(RetentionPolicy.RUNTIME)
@Target(ElementType.TYPE)
public @interface RequestMappingByHmm {
    String value();
}

接口层

@RequestMappingByHmm("/user")
public class UserController {
    @GetMappingByHmm("/queryUserName")
    public String queryUserName(String id) {
        return "hmm:" + id;
    }
    @GetMappingByHmm("/queryUserXX")
    public String queryUserXX() {
        return "xxx";
    }
}

请求实体类

@Data
public class ReqParamDto implements Serializable {
    private String applicationName;//微服务名 单台机器也可能部署了其它用用  客户端传的
    private String url;//接口地址
    private Object reqDto;//请求对象
}

服务发现（没有暴露出去）

/**
 * @author heian
 * @create 2020-03-29-10:44 上午
 * @description 加载所有的路径
 */
public class ControllerLoader {
    //所有接口层{"/user"/queryUserName",controller层的类}
    static Map<String,DetailsReqDto> map = new HashMap<>();
    /**
     * 路径是唯一的可以作为map的key
     * @param pkgName com.example.customers.standard.rpc.api
     */
    public static Map<String,DetailsReqDto> getController(String pkgName) throws Exception {
        List<Class> classes = getClasses(pkgName);//com.example.customers.standard.rpc.api.UserController
        for(Class<?> cla : classes) {
            RequestMappingByHmm annotation = cla.getAnnotation(RequestMappingByHmm.class);
            String path1 = annotation.value();
            Method[] declaredMethods = cla.getDeclaredMethods();
            for (Method declaredMethod : declaredMethods) {
                GetMappingByHmm annotation1 = declaredMethod.getAnnotation(GetMappingByHmm.class);
                if (annotation1 != null){
                    String path2 = annotation1.value();
                    DetailsReqDto dto = new DetailsReqDto();
                    dto.setAClass(cla);
                    dto.setMethodName(declaredMethod.getName());
                    map.put(path1+path2,dto);
                }
            }
        }
        //map 可以考虑存放在zk或者其它分布式配置中心
        System.out.println(map.toString());
        return map;
    }
    /**
     * @param pkgName
     * @return 返回 RequestMappingByHmm 注解下所有的类
     * @throws ClassNotFoundException
     */
    public static List<Class> getClasses(String pkgName) throws ClassNotFoundException{
        List<Class> totalClass = new ArrayList<>();
        ClassLoader cld = Thread.currentThread().getContextClassLoader();
        if (cld == null)
            throw new ClassNotFoundException("无法获取到ClassLoader");
        String path = pkgName.replace('.', '/');
        URL resource = cld.getResource(path); //会去target 下面寻找的编译器文件
        if (resource == null)
            throw new ClassNotFoundException("没有这样的资源：" + path);
        File directory = new File(resource.getFile());
        if (directory.exists()){
            String[] fileName = directory.list();//目录名
            File[] files = directory.listFiles();
            for (int i = 0; i < files.length; i++) {
                File file = files[i];
                //判断是不是比以class 后缀结尾的
                if (file.isFile() && file.getName().endsWith(".class")){
                    String classPath = pkgName + "." + fileName[i].substring(0,fileName[i].length()-6);
                    Class<?> aClass = Class.forName(classPath);
                    RequestMappingByHmm annotation = aClass.getAnnotation(RequestMappingByHmm.class);
                    if (annotation != null){
                        totalClass.add(aClass);
                    }
                }else {
                    //目录则递归查找
                    List<Class> classes1 = getClasses(pkgName + "." + file.getName());
                    if (classes1 != null && !classes1.isEmpty()){
                        totalClass.addAll(classes1);
                    }
                }
            }
        }else {
            throw new ClassNotFoundException(pkgName + "不是一个有效的包");
        }
        return totalClass;
    }
}

大致流程

扫描包下所有带 RequestMapping注解的类，拿到后对应的接口类
获取对应类请求的value1，然后再获取带有 GetMappingBy 注解的方法，获取对应的注解的value2
然后将value1+value2作为map的key，方法名和对应class（当然还有微服务名，这里暂时没用）整合成一个对象作为一个map的value存起来
服务分发：服务端开启，处于侦听状态，一旦接收到客户端的请求，拿到传来的参数，获取缓存map，有了类有了方法，直接method.invoke（类对象，传来的参数）这样就相等于调用了接口一样，这不过我这里是通过socket来简单实现一下。

服务分发

/**
 * @author heian
 * @create 2020-03-29-12:49 上午
 * @description 找到接口请求过来的对象，解析->调用对应的方法
 */
public class ServiceDispatch {
    public static Object dispatch(Object reqObj){
        ReqParamDto reqParamDto = (ReqParamDto) reqObj;
        //String applicationName = reqParamDto.getApplicationName();//暂时 没用到
        String url = reqParamDto.getUrl();
        Object reqDto = reqParamDto.getReqDto();//requestbody
        Object ret = null;
        //根据客户端发来的请求  地址存在则表明存在对应的方法
        if (ControllerLoader.map.get(url) != null){
            DetailsReqDto dto = ControllerLoader.map.get(url);
            Class aClass = dto.getAClass();
            String methodName = dto.getMethodName();
            try {
                Method method = aClass.getMethod(methodName, reqDto.getClass());
                System.out.println("服务端开始执行方法：" + methodName + "，参数类型：" + reqDto.getClass());
                ret = method.invoke(aClass.newInstance(),reqDto);
                System.out.println("执行结果：" + ret.toString());
            }catch (Exception e){
                e.printStackTrace();
            }
        }
        return ret;
    }
}

服务段自身存储的实体类用来标识：接口与方法的映射

/**
 * @author heian
 * @create 2020-03-29-1:34 上午
 * @description
 */
public class BootStrap {
    public static void main(String[] args) throws Exception {
        //加载所有的接口
        ControllerLoader.getController("com.example.customers.standard.rpc.api");
        BioSocketServer.start(8888);
    }
}

服务端启动类

@Setter
@Getter
@ToString
public class DetailsReqDto {
    //通过接口路径  找到对应的类  方法名
    private Class aClass;
    private String methodName;
}

/**
 * @author heian
 * @create 2020-03-29-12:37 上午
 * @description
 */
public class BioSocketServer {
    private static ExecutorService executorService = Executors.newCachedThreadPool();
    public static void start(int port) throws IOException {
        ServerSocket serverSocket = new ServerSocket(port);
        System.out.println("服务端正在启动");
        while (true){
            Socket socket = serverSocket.accept();//阻塞等待客户端连接
            //每一个连接连接进来必须分配一个线程，防止因为某个客户端阻塞而导致无法接受新的请求
            executorService.submit(() -> {
                ObjectInputStream in = null;
                ObjectOutputStream out = null;
                System.out.println("收到客户端的连接进来");
                try {
                    in = new ObjectInputStream(socket.getInputStream());
                    //拿到客户端发来的对象
                    Object reqObj = in.readObject();
                    System.out.println("reqObj:" + reqObj.toString());
                    //调用服务的分发  去查询对应服务
                    Object retObj = ServiceDispatch.dispatch(reqObj);
                    out = new ObjectOutputStream(socket.getOutputStream());
                    //返回客户端接口
                    out.writeObject(retObj);
                    out.flush();
                }catch (Exception e){
                    e.printStackTrace();
                }finally {
                    try {
                        if (in != null) in.close();
                        if (out != null)out.close();
                    }catch (Exception e){
                        e.printStackTrace();
                    }
                }
            });
        }
    }
}

客户端启动类

/**
 * @author heian
 * @create 2020-03-29-12:06 下午
 * @description
 */
public class BioSocketClient {
    public static void main(String[] a) throws IOException {
        SocketAddress socketAddress = new InetSocketAddress("localhost",8888);
        Socket socket = new Socket();
        socket.connect(socketAddress);
        //调用服务
        ReqParamDto reqParamDto = new ReqParamDto();
        reqParamDto.setApplicationName("lcloud-ceres-rnp");
        reqParamDto.setUrl("/user/queryUserName");
        reqParamDto.setReqDto("1");
        ObjectInputStream in = null;
        ObjectOutputStream out = null;
        try {
            out = new ObjectOutputStream(socket.getOutputStream());
            out.writeObject(reqParamDto);
            out.flush();
            in = new ObjectInputStream(socket.getInputStream());
            Object retObj = in.readObject();
            System.out.println("收到服务段的返回信息" + retObj.toString());
        }catch (Exception e){
            e.printStackTrace();
        }finally {
            try {
                if (in != null) in.close();
                if (out != null)out.close();
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    }
}

屏幕快照 2020-03-29 下午1.27.00.png

测试

分别启动服务段和客户端
server

服务端正在启动
收到客户端的连接进来
reqObj:ReqParamDto(applicationName=lcloud-ceres-rnp, url=/user/queryUserName, reqDto=1)
服务端开始执行方法：queryUserName，参数类型：class java.lang.String
执行结果：hmm:1

client

收到服务段的返回信息hmm:1

第五章 手写RPC框架

前请提示

代码演示

自定义注解仿照spring

接口层

请求实体类

服务发现（没有暴露出去）

大致流程

服务分发

服务段自身存储的实体类用来标识：接口与方法的映射

服务端启动类

客户端启动类

测试