轻量级RPC设计与实现第二版

Posted 2021-03-09 maratong

在上一个版本中利用netty实现了简单的一对一的RPC，需要手动设置服务地址，限制性较大。
在本文中，利用zookeeper作为服务注册中心，在服务端启动时将本地的服务信息注册到zookeeper中，当客户端发起远程服务调用时，先从zookeeper中获取该服务的地址，然后根据获得的这个地址来利用netty进行网络传送。
在服务端和注册中心之间需要建立监听，当服务信息发生变化或网络连接等问题时需要对注册中心的服务信息进行修改。在本文中创建了服务注册监控中心，利用心跳机制来判断与服务端是否有较稳定的连接，当出现网络不稳定时，则从注册中心中删除属于该服务端的服务信息。在本项目中设定在5分钟内3次以上没有发送心跳包为不稳定状态。

关于心跳机制，之前有一篇文章介绍过：Dubbo心跳机制

zookeeper注册中心

zookeeper是hadoop中的一个重要组件，其主要是作为分布式协调服务
zookeeper采用节点树的数据模型，类似linux文件系统。
每个节点称做一个ZNode,每个ZNode都可以通过路径唯一标识，同时每个节点还可以存储少量数据。
本项目借鉴dubbo的注册中心模型来设计本文的注册中心。
总体上设计了四级节点，在一个节点是一个持久节点/register，表示是记录注册服务的区域。二级节点是服务接口名，三级节点是远程服务ip地址，该节点是临时节点，节点存储的数据是具体的实现类名。

在客户端会根据服务接口名在注册中心进行查找，得到远程服务ip地址，并根据节点中存储的具体实现类名进行反射。

首先进行zookeeper初始化,利用了CuratorFramework有关类

private static void init() {
        RetryPolicy retryPolicy = new RetryNTimes(ZKConsts.RETRYTIME, ZKConsts.SLEEP_MS_BEWTEENR_RETRY);
        client = CuratorFrameworkFactory.builder().connectString(ZKConsts.ZK_SERVER_PATH)
                .sessionTimeoutMs(ZKConsts.SESSION_TIMEOUT_MS).retryPolicy(retryPolicy)
                .namespace(ZKConsts.WORK_SPACE).build();
        client.start();
    }

服务的注册代码

public static void register(URL url) {
        try {
            String interfaceName = url.getInterfaceName();
            String implClassName = url.getImplClassName();
            Stat stat = client.checkExists().forPath(getPath(interfaceName, url.toString()));
            if (stat != null) {
                System.out.println("该节点已存在！");
                client.delete().forPath(getPath(interfaceName, url.toString()));
            }
            client.create()
                    .creatingParentsIfNeeded()
                    .withMode(CreateMode.EPHEMERAL)
                    //权限控制，任何连接的客户端都可以操作该属性znode
                    .withACL(ZooDefs.Ids.OPEN_ACL_UNSAFE)
                    .forPath(getPath(interfaceName, url.toString()), implClassName.getBytes());
            System.out.println(getPath(interfaceName, url.toString()));
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

根据服务接口名来获取远程服务连接地址

public static URL random(String interfaceName) {
        try {

            System.out.println("开始查找服务节点：" + getPath(interfaceName));
            List<String> urlList = client.getChildren().forPath("/" + interfaceName);
            System.out.println("结果：" + urlList);
            String serviceUrl = urlList.get(0);
            String[] urls = serviceUrl.split(":");
            String implClassName = get(interfaceName, serviceUrl);
            return new URL(urls[0], Integer.valueOf(urls[1]), interfaceName, implClassName);
        } catch (Exception e) {
            System.out.println(e);
            e.printStackTrace();
        }
        return null;
    }

注册中心与服务端进行连接时需要判断是否维持了稳定的连接，如果服务端出现宕机等情况时需要从注册中心中删除这些服务。
以前的一些处理机制，有session机制和wacher机制。
session机制
每个zookeeper注册中心与服务端进行连接时会创建一个session,在设置的sessionTimeout内，服务端会与注册中心进行心跳包的定时发送，从而感知每个客户端是否宕机，如果创建某个临时Znode节点对应的session销毁时，相应的临时节点也会被注册中心删除。
watcher机制
针对每个节点的操作，都有要给监督者进行watcher,当监控的某个节点发生了变化，则会触发watcher事件。注册中心的watcher是一次性的，触发后会被销毁。父节点，子节点增删改都能够触发watcher。触发销毁后，下次需要监听时还需要再注册一次。
本文心跳机制
服务端定时向注册中心发送本机地址，看作心跳数据包，而注册中心监控则维持一个channelId和具体地址的map,并且通过IdleHandler监听空闲事件，到达一定的空闲次数则认为不活跃，当不活跃时，zookeeper删除对应的url节点。该版本实现了上面的内容，后续的步骤在以后的版本实现。
如果10s内没有触发读，就会执行userEventTriggered方法。如果5分钟中出现两次不活跃次数，就认定该连接不稳定，注册中心会移除属于该服务端的服务。你也可以根据实际情况设定不稳定标准。

 service.scheduleAtFixedRate(() -> {
                if (future.channel().isActive()) {
                    int time = new Random().nextInt(5);
                    log.info("本次定时任务获取的随机数:{}", time);
                    if (time > 3) {
                        log.info("发送本地地址到注册中心：{}", url);
                        future.channel().writeAndFlush(url);
                    }
                }
            }, 60, 60, TimeUnit.SECONDS);

 @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent state = (IdleStateEvent)evt;
            if (state.state().equals(IdleState.READER_IDLE)) {
                log.info("读空闲");
            } else if (state.state().equals(IdleState.WRITER_IDLE)) {
                log.info("写空闲");
            }
            //在一定时间内读写空闲才会关闭链接
            else if (state.state().equals(IdleState.ALL_IDLE)) {
                if (++inActiveCount == 1) {
                    start = System.currentTimeMillis();
                }
                int minute = (int)((System.currentTimeMillis() - start) / (60 * 1000)) + 1;
                log.info("第{}次读写都空闲，计时分钟数{}", inActiveCount, minute);
                if (inActiveCount > 2 && minute < 5) {
                    log.info("移除不活跃ip");
                    removeAndClose(ctx);
                } else {
                    if (minute >= 5) {
                        log.info("新周期开始");
                        start = 0;
                        inActiveCount = 0;
                    }
                }
            }
        }
    }

具体实现代码：RPC第二版