手写通俗简版RPC通信框架，你上你也行！

àì夳堔傛蜴生んèń 2021-11-17 09:12 251阅读 0赞

# 背景 #

上篇文章我简单的介绍了自己打造的通俗简版RPC通信框架，也得到了大家的支持，短短一个星期的阅读量就上千了，虽然之后涨得很慢，哈哈，但是也很感动~

如果大家对此项目还感兴趣的话，可到GitHub上瞄瞄：[Netty-RPC][]

上篇文章链接：[《SpringBoot2+Netty打造通俗简版RPC通信框架》][SpringBoot2_Netty_RPC]

在介绍后续新增功能前，我们先回顾一下最简单的RPC通信的流程，流程图如下：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0hvd2luZnVu_size_16_color_FFFFFF_t_70][]

我们可以看到其实整个流程其实是非常的简单的：客户端接收前端发送的请求，封装好请求Packet根据配置打开Netty通道进行通信，服务端接收请求Packet，解析并且根据请求信息，反射获取实现类调用方法，得到结果并封装好响应Packet然后返回结果给客户端，客户端获取结果响应给前端。

# 新增功能 #

因为是模仿Dubbo造RPC通信框架，那么Dubbo基本的功能我们当然也必须得有啦，下面列出的是我后续新增的优化：

1.  单一长连接
2.  服务注解，并且带版本号
3.  增加注册中心
4.  处理Netty客户端或服务端主动断开异常
5.  业务处理使用自定义业务线程池

详细的代码大家可到我的github上阅读。

# 详细介绍 #

首先，我们先看一下带注册中心后的流程图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0hvd2luZnVu_size_16_color_FFFFFF_t_70 1][]

我们可以看到多了个Zookeeper作为注册中心，然后就到了监听缓存列表，和服务缓存列表。大家不理解这两个列表不重要，下面我将继续详细的介绍一下我做这些功能的思路。

### 1、单一长连接： ###

首先上一下简单的流程图：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0hvd2luZnVu_size_16_color_FFFFFF_t_70 2][]

思路非常的简单：就是使用内存缓存缓存起来，结构就是Map，key为IP:Port  value为channel。

在没有注册中心时，服务端的IP和PORT是写在配置文件里的，我们直接获取IP信息，并且判断在【Channel缓存列表】是否有此IP对应缓存起来的的Channel，有则获取直接进行RPC通信，否则创建新Channel进行RPC通信，记得还要缓存起来。  
        而在有注册中心时，我们需要在请求Packet中获取需要提供的服务名，然后根据服务名在【服务缓存列表】获取所有提供此服务的所有应用IP，然后就是判断IP列表在【Channel缓存列表】中是否有缓存的Channel，有则获取并直接进行RPC通信；  
否则，遍历应用IP列表，直到创建新Channel通信并且连接成功，然后将Channel缓存起来。最后，进行RPC通信然后等待获取Result即可。

### 2、服务注解： ###

我们知道需要进行RPC通信的接口都会加上@NettyRPC注解，然后在服务端这，每次都是使用Reflections框架扫描出指定路径下的所有类，再判断是否有服务的实现类，有则利用反射进行方法调用。这听起来可是相当浪费时间，哈哈。

这时候，我们可以利用Spring框架来去除掉这个扫描的动作。首先，我们提供一个@NettyRpcService注解，来标识所有RPC服务的实现类。然后我们创建一个类，实现接口ApplicationContextAware。然后我们可以利用ApplicationContext的getBeansWithAnnotation方法类根据指定注解获取Bean，我们这里当然是指定前面的@NettyRPCService注解了，然后我们利用内存缓存来缓存起来【提供RPC服务的实现类列表】，结构为Map：key为接口名+版本号，value为bean实例。那么之后的大家应该就懂了，我们再也不必每次都使用Reflections框架了。

package com.hyf.rpc.netty.server.config;
    
    import com.hyf.rpc.netty.anno.NettyRPCService;
    import com.hyf.rpc.netty.properties.NettyProperties;
    import com.hyf.rpc.netty.server.NettyServer;
    import org.springframework.beans.BeansException;
    import org.springframework.beans.factory.annotation.Autowired;
    import org.springframework.context.ApplicationContext;
    import org.springframework.context.ApplicationContextAware;
    import org.springframework.context.ApplicationListener;
    import org.springframework.context.event.ContextRefreshedEvent;
    import org.springframework.stereotype.Component;
    
    import java.util.HashMap;
    import java.util.Map;
    
    /**
     * @author Howinfun
     * @desc Netty服务提供者启动&扫面存储提供服务的实现类
     * @date 2019/7/18
     */
    @Component
    public class NettyServerInitConfig implements ApplicationListener<ContextRefreshedEvent>, ApplicationContextAware {
    
        /** 提供RPC服务的实现类  key为接口名+版本号，value为bean实例*/
        public static final Map<String,Object> beanMap = new HashMap<>(10);
    
        @Autowired
        private NettyServer nettyServer;
        @Autowired
        private NettyProperties nettyProperties;
        /**
         * 当ApplicationContext初始或刷新完毕触发
         * @param event
         */
        @Override
        public void onApplicationEvent(ContextRefreshedEvent event) {
            if (nettyProperties.getServerPort() != null){
                nettyServer.start();
            }
        }
    
        @Override
        public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
            // 遍历带有NettyRPCService注释的服务实现类
            Map<String,Object> beans = applicationContext.getBeansWithAnnotation(NettyRPCService.class);
            if (beans != null && beans.size() > 0) {
                for (Object serviceBean : beans.values()) {
                    String interfaceName = serviceBean.getClass().getAnnotation(NettyRPCService.class).value().getName();
                    String version = serviceBean.getClass().getAnnotation(NettyRPCService.class).version();
                    beanMap.put(interfaceName+version, serviceBean);
                }
            }
            System.out.println(beanMap.toString());
    
        }
    }

### 3、注册中心： ###

这个上一下简单的流程图先：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0hvd2luZnVu_size_16_color_FFFFFF_t_70 3][]

顺便看一下Zookeeper的数据结构：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0hvd2luZnVu_size_16_color_FFFFFF_t_70 4][]

我这里使用的是Zookeeper作为注册中心。首先大家得自己安装一个Zookeeper服务。我们做注册中心是利用Zookeeper的监听事件，当然了，Zookeeper原生的监听事件是利用Watcher，而且是一次性的，所以不用。我会使用第三方框架Curator。Curator引入了Cache来实现对Zookeeper服务端事件监听，Cache事件监听可以理解为一个本地缓存视图与远程Zookeeper视图的对比过程。Cache提供了反复注册的功能。Cache分为两类注册类型：节点监听和子节点监听。

首先是服务端：在zookeeper的/root下创建名为配置文件中namespace(例如rpc-server)的节点，然后我们可在上面的扫描带@NettyRPCService实现类缓存起来的同时，每一个实现类就往/root/namespace下创建一个节点，节点名为@NettyRPCService的value().getName+version()，节点内容为IPPojo(ip+提供netty服务端口号)的Json字符串。  
        再来客户端：客户端主要是要做监听，首先是监听(使用PathChildrenCache,可监听子节点的增删改)根节点/root，如果有节点新增，则表明有新应用提供服务：这时候我们又要继续对此节点做监听(也是使用PathChildrenCache)，并且将PathChildrenCache缓存到内存缓存中【监听缓存列表】。如果有节点删除，则表明有应用不提供服务了：这时候我们将监听关闭掉，并且从【监听缓存列表】中删除，并且从【服务缓存列表】删除应用提供的所有服务。根节点/root下面的子节点才是真正提供的RPC服务：当新增，需要将节点信息缓存到内存缓存中【服务缓存列表】；当删除，从【服务缓存列表】中删除对应数据。

### 4、处理Netty客户端或服务端主动异常： ###

客户端：在RPCResponsPacketHandler中重写exceptionCaught方法。首先是根据标识从【Channel缓存列表】中移除此Channel，然后根据标识从【服务缓存列表】中移除对应的所有服务，最后关闭通道ctx.channel().close()。  
        服务端：直接关闭通道即可，ctx.channel().close();

### 5、业务处理使用自定义业务线程池：      ###

首先自定义一个线程池，根据自己需求去设计 核心线程数、最大线程数、线程保持活跃时间、队列、拒绝策略等。然后在业务处理除直接新建一个任务(实现Runnable接口)提交到线程池处理即可。

package com.hyf.rpc.netty.common;
    
    import lombok.extern.slf4j.Slf4j;
    
    import java.util.concurrent.Future;
    import java.util.concurrent.LinkedBlockingQueue;
    import java.util.concurrent.ThreadPoolExecutor;
    import java.util.concurrent.TimeUnit;
    
    /**
     * @author Howinfun
     * @desc
     * @date 2019/7/12
     */
    @Slf4j
    public class TaskThreadPool {
    
        public static final TaskThreadPool INSTANCE  = new TaskThreadPool();
        private final ThreadPoolExecutor executor;
        private TaskThreadPool(){
            /**
             * 核心线程数：10
             * 最大线程数：20
             * 线程保持活跃时间：60s
             * 队列：阻塞队列，最多存放100个任务
             * 拒绝策略：任务将被放弃
             */
            this.executor = new ThreadPoolExecutor(10,
                                                20,
                                                60, TimeUnit.SECONDS,
                    new LinkedBlockingQueue<>(100),new ThreadPoolExecutor.CallerRunsPolicy());
        }
        public Future submit(Runnable task){
            log.info("业务线程池执行任务中...");
            Future future = executor.submit(task);
            return future;
        }
    }

### 最后： ###

到这里就基本已经全部结束了，虽然总体做得不是很好，但是起码自己经历了从0到1的所谓手写框架了，哈哈哈。同时，也将学到的Netty和Zookeeper等技术稍微实战了一下子，算是很满足了。也感谢很多同学们对我文章的阅读和支持，非常感谢。接下来，得想想干点什么。。。。

[Netty-RPC]: https://github.com/Howinfun/Netty-RPC
[SpringBoot2_Netty_RPC]: https://blog.csdn.net/Howinfun/article/details/96481503
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0hvd2luZnVu_size_16_color_FFFFFF_t_70]: /images/20211116/be9f494acea74836bfad7ee29022435a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0hvd2luZnVu_size_16_color_FFFFFF_t_70 1]: /images/20211116/dc77cc278925425183af198ca33d1996.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0hvd2luZnVu_size_16_color_FFFFFF_t_70 2]: /images/20211116/4a21275d67c84b53af178949fc5066e7.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0hvd2luZnVu_size_16_color_FFFFFF_t_70 3]: /images/20211116/2abc826d846c4e71bb176d75f741b085.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0hvd2luZnVu_size_16_color_FFFFFF_t_70 4]: /images/20211116/07e30104a51e462f9d070733507b4bcf.png