Nacos源码解读12——Nacos中长连接的实现

news2024/11/25 16:34:12

短连接 VS 长连接

什么是短连接

客户端和服务器每进行一次HTTP操作，就建立一次连接，任务结束就中断连接。

长连接

客户端和服务器之间用于传输HTTP数据的TCP连接不会关闭，客户端再次访问这个服务器时，会继续使用这一条已经建立的连接。

长连接的好处

假如请求一个普通的网页但是这个网页有很多个 css js请求那每次打开一个网页，基本就要建立几个甚至几十个TCP连接，浪费很多网络资源。如果是长连接的话，那么这么多HTTP请求（包括请求网页的内容、CSS文件、JS文件、图片等）都是使用的一个TCP连接，显然可以节省很多资源。

另外一点，长连接并不是永久连接的。如果一段时间内（具体时间可以在header中进行设置，也就是所谓的超时时间），这个连接没有HTTP请求发出的话，那么这个长连接就会被断掉。

短轮询VS 长轮询

什么是轮询

就比如在飞机上想上厕所却发现厕所有人需要等待但是你又不想在那等就只能回去等，但是过一段时间在去还有人在接着回去这么周而复始的去回

短轮询

轮询的原理就是客户端以一定的时间间隔向服务端发出请求，频繁的请求保持客户端和服务端同步。
短轮询最大的问题就是客户端发出请求和服务器端的更新并不是一致的。客户端以固定的频率想服务器发出请求，可能服务器端并没有更新，返回的是个空的信息，等服务器端更新的时候，有可能客户端并没有请求，而且只有最后一次请求才能获得最新数据，这样多次请求不仅浪费了资源，而且并不是实际上的实时更新。

优缺点

优点：后端程序编写比较容易
缺点：请求中有大半是无用，浪费带宽和服务器资源。（而每一次的 HTTP 请求和应答都带有完整的 HTTP 头信息，这就增加了每次传输的数据量）

长轮询

页面发起一个到服务器的请求，然后服务器一直保持连接打开，直到有数据可以发送。
发送完数据之后，浏览器关闭连接，随即又发送一个到服务器的新请求。这一过程在页面打开期间一直持续不断。（在服务端hold住Http请求（死循环或者sleep等等方式），等到目标时间发生，返回Http响应。

长轮询优缺点

优点：在无消息的情况下不会频繁的请求，节省了网络流量，解决了服务端一直疲于接受请求的窘境。
缺点：服务器hold连接会消耗资源，需要同时维护多个线程，服务器所能承载的TCP连接数是有上限的，这种轮询很容易把连接数顶满。

Nacos是怎么处理服务配置修改刷新

1.X版本

在这里插入图片描述
客户端启动的时候会初始化一个长连接的线程池定时去发送pull请求会发送一个HTTP请求到服务端在服务端hold住Http请求超时时间是30秒
30s内有可能触发变化 push数据到客户端
30s内数据没有变更超时返回

缺点

30 秒定期创建销毁连接，GC压力大

2.X版本

Nacos 2.x 相比上面 30s ⼀次的长轮训，升级成长链接模式，配置变更，启动建立长链接，配置变
更服务端推送变更配置列表，然后 SDK 拉取配置更新，因此通信效率大幅提升

Nacos长连接源码解析

这里只抓重点想看详情的话看前面几章

客户端变更推送配置变更事件

    /**
     * adaptor to config module ,when server side config change ,invoke this method.
     *
     * @param groupKey groupKey
     */
    public void configDataChanged(String groupKey, String dataId, String group, String tenant, boolean isBeta,
            List<String> betaIps, String tag) {
        //获取注册的Client列表
        Set<String> listeners = configChangeListenContext.getListeners(groupKey);
        if (CollectionUtils.isEmpty(listeners)) {
            return;
        }
        int notifyClientCount = 0;
        //遍历client列表
        for (final String client : listeners) {
            //拿到grpc连接
            Connection connection = connectionManager.getConnection(client);
            if (connection == null) {
                continue;
            }
            
            ConnectionMeta metaInfo = connection.getMetaInfo();
            //beta ips check.
            String clientIp = metaInfo.getClientIp();
            String clientTag = metaInfo.getTag();
            if (isBeta && betaIps != null && !betaIps.contains(clientIp)) {
                continue;
            }
            //tag check
            if (StringUtils.isNotBlank(tag) && !tag.equals(clientTag)) {
                continue;
            }
            //构建请求参数
            ConfigChangeNotifyRequest notifyRequest = ConfigChangeNotifyRequest.build(dataId, group, tenant);
            //构建推送任务
            RpcPushTask rpcPushRetryTask = new RpcPushTask(notifyRequest, 50, client, clientIp, metaInfo.getAppName());
            //推送任务 向客户端发送变更通知
            push(rpcPushRetryTask);
            notifyClientCount++;
        }
        Loggers.REMOTE_PUSH.info("push [{}] clients ,groupKey=[{}]", notifyClientCount, groupKey);
    }

ClientWorker 这里最终会调用 notifyListenConfig方法这个方法实际上就是往listenExecutebell这个阻塞队列中去offer一个object对象

 private void initRpcClientHandler(final RpcClient rpcClientInner) {
             rpcClientInner.registerServerRequestHandler((request) -> {
                if (request instanceof ConfigChangeNotifyRequest) {
                    ConfigChangeNotifyRequest configChangeNotifyRequest = (ConfigChangeNotifyRequest) request;
                    LOGGER.info("[{}] [server-push] config changed. dataId={}, group={},tenant={}",
                            rpcClientInner.getName(), configChangeNotifyRequest.getDataId(),
                            configChangeNotifyRequest.getGroup(), configChangeNotifyRequest.getTenant());
                    String groupKey = GroupKey
                            .getKeyTenant(configChangeNotifyRequest.getDataId(), configChangeNotifyRequest.getGroup(),
                                    configChangeNotifyRequest.getTenant());
                    
                    CacheData cacheData = cacheMap.get().get(groupKey);
                    if (cacheData != null) {
                        synchronized (cacheData) {
                            cacheData.getLastModifiedTs().set(System.currentTimeMillis());
                            cacheData.setSyncWithServer(false);
                            //向阻塞队列中添加元素 触发长连接的执行
                            notifyListenConfig();
                        }
                        
                    }
                    //返回客户端响应
                    return new ConfigChangeNotifyResponse();
                }
                return null;
            });

}

客户端处理变更事件

客户端启动的时候会创建 NaocsConfigService 他的构造方法中会创建一个ClientWorker并启动然后会执行 startInternal方法这个方法中会看到从listenExecutebell中拿数据 listenExecutebell在上文服务配置发生变更的时候会往里面塞一个object 对象所以这里就能poll队列中出来如果队列为空等待5秒后执行，如果队列不为空立即执行

        @Override
        public void startInternal() {
            executor.schedule(() -> {
                while (!executor.isShutdown() && !executor.isTerminated()) {
                    try {
                        listenExecutebell.poll(5L, TimeUnit.SECONDS);
                        if (executor.isShutdown() || executor.isTerminated()) {
                            continue;
                        }
                        executeConfigListen();
                    } catch (Exception e) {
                        LOGGER.error("[ rpc listen execute ] [rpc listen] exception", e);
                    }
                }
            }, 0L, TimeUnit.MILLISECONDS);
            
        }

executeConfigListen 方法详细看 https://blog.csdn.net/qq_41956309/article/details/134904263这段下面简述一下
其实就是将3000个配置信息封装成一个CacheData 共用一个TaskId 一个TaskId 对应一个Client连接然后会发送到服务端通过比较md5的方式来判断哪些配置发生了变更然后会返回变更的key 然后客户端遍历返回的有变更信息的key的信息在去调用服务端查找具体的配置信息返回客户端然后做返回和动态刷新以及本地缓存的修改
在这里插入图片描述