Nacos 服务注册服务端源码分析（五）

港控/mmm° 2023-09-25 00:12 92阅读 0赞

## Task分析 ##

NacosTask只有一个方法boolean shouldProcess()，即判断是否应该执行。

它有有两个抽象的子类，分别是AbstractExecuteTask和AbstractDelayTask

![format_png][]

AbstractExecuteTask实现很简单，就是为true需要执行。

@OverridepublicbooleanshouldProcess() {
        returntrue;
    }
    复制代码

AbstractDelayTask是延迟执行的，所以它有个延迟机制的判断。

@OverridepublicbooleanshouldProcess() {
        // 如果当前时间减去上次时间还未到达任务执行的间隔时间则返回falsereturn (System.currentTimeMillis() - this.lastProcessTime >= this.taskInterval);
    }
    // 抽象类，由子类判断添加的任务是否可以合并publicabstractvoidmerge(AbstractDelayTask task);
    复制代码

这里还是用上篇分析的PushDelayTask分析下这个方法的作用

@Overridepublicvoidmerge(AbstractDelayTask task) {
        if (!(task instanceof PushDelayTask)) {
            return;
        }
        PushDelayTaskoldTask= (PushDelayTask) task;
        if (isPushToAll() || oldTask.isPushToAll()) {
            // 可以push到所有的的这一类，无需特殊处理
            pushToAll = true;
            targetClients = null;
        } else {
            // 设置个集合，将连接放入，后续要一个个的推，这部分是针对失败的数据
            targetClients.addAll(oldTask.getTargetClients());
        }
        setLastProcessTime(Math.min(getLastProcessTime(), task.getLastProcessTime()));
        Loggers.PUSH.info("[PUSH] Task merge for {}", service);
    }
    复制代码

这里也和之前分析的NacosDelayTaskExecuteEngine和NacosExecuteTaskExecuteEngine对应。因为NacosExecuteTaskExecuteEngine无需延迟，所以默认就是需要执行，来一个执行一个。而NacosDelayTaskExecuteEngine则会有个延迟的处理，主要针对失败的，延迟处理。因为既然是失败的，100ms对于失败的处理来说，间隔时间还是短了一些，所以放入到集合中，等延迟时间到了后再统一处理，这里的延迟时间的1s。

下面我们继续上次的代码往后分析

delayTaskEngine.getPushExecutor().doPushWithCallback(each, subscriber, wrapper,
                            newServicePushCallback(each, subscriber, wrapper.getOriginalData(), delayTask.isPushToAll()));
    复制代码

第一步是获取delayTaskEngine.getPushExecutor()，跟踪这个类分析，发现是由构造方法传入的。再往上跟踪，可以看到该类是归spring托管的。注入的PushExecutorDelegate。

这个Delegate单词翻译是委托，说明这个是委托模式。看下这个类的一些成员变量和核心方法。

// rpc执行类，V2版本使用privatefinal PushExecutorRpcImpl rpcPushExecuteService;
    // udp执行类，V1版本使用privatefinal PushExecutorUdpImpl udpPushExecuteService;
    
    publicPushExecutorDelegate(PushExecutorRpcImpl rpcPushExecuteService, PushExecutorUdpImpl udpPushExecuteService) {
        this.rpcPushExecuteService = rpcPushExecuteService;
        this.udpPushExecuteService = udpPushExecuteService;
    }
    
    private PushExecutor getPushExecuteService(String clientId, Subscriber subscriber) {
        Optional<SpiPushExecutor> result = SpiImplPushExecutorHolder.getInstance()
            .findPushExecutorSpiImpl(clientId, subscriber);
        if (result.isPresent()) {
            return result.get();
        }
        // use nacos default push executor// 根据连接的客户端id识别是由upd推送还是rpc推送return clientId.contains(IpPortBasedClient.ID_DELIMITER) ? udpPushExecuteService : rpcPushExecuteService;
    }
    复制代码

本篇分析的2.2.0的源码，自然是rpc版本的推送。至于为什么弃用upd，大家可以看下第一篇的 [1、Nacos 服务注册客户端源码分析][1_Nacos]中的引文，这里不做解释了。

我们顺着思路继续分析com.alibaba.nacos.naming.push.v2.executor.PushExecutorRpcImpl\#doPushWithCallback

@OverridepublicvoiddoPushWithCallback(String clientId, Subscriber subscriber, PushDataWrapper data,
                                   NamingPushCallback callBack) {
        // 获取服务信息ServiceInfoactualServiceInfo= getServiceInfo(data, subscriber);
        callBack.setActualServiceInfo(actualServiceInfo);
        // 构建一个NotifySubscriberRequest，通过grpc向客户端发送信息
        pushService.pushWithCallback(clientId, NotifySubscriberRequest.buildNotifySubscriberRequest(actualServiceInfo),
                                     callBack, GlobalExecutor.getCallbackExecutor());
    }
    
    publicvoidpushWithCallback(String connectionId, ServerRequest request, PushCallBack requestCallBack,
                                 Executor executor) {
        // 拿到客户端的连接Connectionconnection= connectionManager.getConnection(connectionId);
        if (connection != null) {
            try {
                // 发送异步请求
                connection.asyncRequest(request, newAbstractRequestCallBack(requestCallBack.getTimeout()) {
    
                    @Overridepublic Executor getExecutor() {
                        return executor;
                    }
    
                    @OverridepublicvoidonResponse(Response response) {
                        if (response.isSuccess()) {
                            requestCallBack.onSuccess();
                        } else {
                            requestCallBack.onFail(newNacosException(response.getErrorCode(), response.getMessage()));
                        }
                    }
    
                    @OverridepublicvoidonException(Throwable e) {
                        requestCallBack.onFail(e);
                    }
                });
            } catch (ConnectionAlreadyClosedException e) {
                connectionManager.unregister(connectionId);
                requestCallBack.onSuccess();
            } catch (Exception e) {
                Loggers.REMOTE_DIGEST
                    .error("error to send push response to connectionId ={},push response={}", connectionId,
                           request, e);
                requestCallBack.onFail(e);
            }
        } else {
            requestCallBack.onSuccess();
        }
    }
    复制代码

这里构建了一个NotifySubscriberRequest。回忆一下我们之前分析的代码。一个Request代表着一类网络请求，一定有着一个Handler处理。这里是由com.alibaba.nacos.client.naming.remote.gprc.NamingPushRequestHandler来处理。

可以看到这个包在client包中。也就是grpc的双向流的处理。可以直接接受服务的请求，客户端直接处理。我们简单看下客户端的处理。

publicclassNamingPushRequestHandlerimplementsServerRequestHandler {  
        privatefinal ServiceInfoHolder serviceInfoHolder;
        publicNamingPushRequestHandler(ServiceInfoHolder serviceInfoHolder) {
            this.serviceInfoHolder = serviceInfoHolder;
        }
        
        @Override
        public Response requestReply(Request request) {
            if (request instanceof NotifySubscriberRequest) {
                NotifySubscriberRequestnotifyRequest= (NotifySubscriberRequest) request;
                // 处理逻辑
                serviceInfoHolder.processServiceInfo(notifyRequest.getServiceInfo());
                returnnewNotifySubscriberResponse();
            }
            returnnull;
        }
    }
    
    public ServiceInfo processServiceInfo(ServiceInfo serviceInfo) {
        StringserviceKey= serviceInfo.getKey();
        if (serviceKey == null) {
            returnnull;
        }
        // 获取老的服务
        ServiceInfooldService= serviceInfoMap.get(serviceInfo.getKey());
        if (isEmptyOrErrorPush(serviceInfo)) {
            //empty or error push, just ignore
            return oldService;
        }
        // 放入客户端的缓存
        serviceInfoMap.put(serviceInfo.getKey(), serviceInfo);
        // 对比下是否发生改变
        booleanchanged= isChangedServiceInfo(oldService, serviceInfo);
        if (StringUtils.isBlank(serviceInfo.getJsonFromServer())) {
            serviceInfo.setJsonFromServer(JacksonUtils.toJson(serviceInfo));
        }
        MetricsMonitor.getServiceInfoMapSizeMonitor().set(serviceInfoMap.size());
        if (changed) {
            NAMING_LOGGER.info("current ips:({}) service: {} -> {}", serviceInfo.ipCount(), serviceInfo.getKey(),
                               JacksonUtils.toJson(serviceInfo.getHosts()));
            // 若改变，发送改变事件
            NotifyCenter.publishEvent(newInstancesChangeEvent(notifierEventScope, serviceInfo.getName(), serviceInfo.getGroupName(),
                                                               serviceInfo.getClusters(), serviceInfo.getHosts()));
            // 磁盘缓存也写一份
            DiskCache.write(serviceInfo, cacheDir);
        }
        return serviceInfo;
    }
    复制代码

## 总结 ##

本篇的内容不多，主要是对上篇的一些扫尾的内容。下一篇我将这个注册服务端再整理一遍，毕竟分了几个章节进行讲解，知识点也比较零碎。可以点击下面的链接直接查看。

作者：ruipost

链接：[https://juejin.cn/post/7210375522504212536][https_juejin.cn_post_7210375522504212536]

[format_png]: https://img-blog.csdnimg.cn/img_convert/f105d461ca29ca5153ed9510db52c173.webp?x-oss-process=image/format,png
[1_Nacos]: https://juejin.cn/post/7207702606645952567
[https_juejin.cn_post_7210375522504212536]: https://juejin.cn/post/7210375522504212536