當前位置：首頁 > news >正文

越南做網站服務器最新app推廣項目平臺

news 2025/7/13 14:18:00

越南做網站服務器,最新app推廣項目平臺,現在網站建設還用測瀏覽器嗎,怎么做繁體字網站前言： 前文我們分析了 Nacos 服務發(fā)現（訂閱）的流程，從 Nacos Client 端的源碼分析了服務發(fā)現的過程，服務發(fā)現最終還是要調用 Nacos Server 端來獲取服務信息，緩存到客戶端本地，并且會定時向 Na…

前言：

前文我們分析了 Nacos 服務發(fā)現（訂閱）的流程，從 Nacos Client 端的源碼分析了服務發(fā)現的過程，服務發(fā)現最終還是要調用 Nacos Server 端來獲取服務信息，緩存到客戶端本地，并且會定時向 Nacos Server 端發(fā)送請求，獲取服務信息，本篇我們從 Nacos Server 來分析一下服務訂閱源碼。

Nacos 系列文章傳送門：

Nacos 初步認識和 Nacos 部署細節(jié)

Nacos 配置管理模型 – 命名空間（Namespace）、配置分組（Group）和配置集ID（Data ID）

Nacos 注冊中心和配置中心【實戰(zhàn)】

服務啟動何時觸發(fā) Nacos 的注冊流程？

Nacos Client 端服務注冊流程源碼分析

Nacos Server 端服務注冊流程源碼分析

Nacos 服務發(fā)現（訂閱）源碼分析（客戶端）

InstanceController#list 方法源碼解析

前文我們分析到服務的發(fā)現（訂閱）最終會調用 Nacos Server 端的接口，而這個接口就在 InstanceController 中，根據接口路徑我們找到了對應的方法也就是 InstanceController#list 方法，源碼解析如下：

//com.alibaba.nacos.naming.controllers.InstanceController#list
@GetMapping("/list")
@Secured(parser = NamingResourceParser.class, action = ActionTypes.READ)
public ObjectNode list(HttpServletRequest request) throws Exception {//獲取 namespaceIdString namespaceId = WebUtils.optional(request, CommonParams.NAMESPACE_ID, Constants.DEFAULT_NAMESPACE_ID);//獲取 serviceNameString serviceName = WebUtils.required(request, CommonParams.SERVICE_NAME);//檢查格式NamingUtils.checkServiceNameFormat(serviceName);//agent  java、c、c++、go、nginx、dnsfString agent = WebUtils.getUserAgent(request);//獲取集群信息String clusters = WebUtils.optional(request, "clusters", StringUtils.EMPTY);String clientIP = WebUtils.optional(request, "clientIP", StringUtils.EMPTY);//獲取 udp  端口int udpPort = Integer.parseInt(WebUtils.optional(request, "udpPort", "0"));//獲取環(huán)境信息String env = WebUtils.optional(request, "env", StringUtils.EMPTY);boolean isCheck = Boolean.parseBoolean(WebUtils.optional(request, "isCheck", "false"));String app = WebUtils.optional(request, "app", StringUtils.EMPTY);String tenant = WebUtils.optional(request, "tid", StringUtils.EMPTY);boolean healthyOnly = Boolean.parseBoolean(WebUtils.optional(request, "healthyOnly", "false"));//獲取服務列表return doSrvIpxt(namespaceId, serviceName, agent, clusters, clientIP, udpPort, env, isCheck, app, tenant,healthyOnly);
}

InstanceController#list 方法本身沒有什么難懂邏輯，只是從 request 中獲取一些屬性后，繼續(xù)調用了 InstanceController#doSrvIpxt 方法，我們接著往下看。

InstanceController#doSrvIpxt 方法源碼解析

InstanceController#doSrvIpxt 方法的源碼比較多，大概拆分一下重要步驟，做了如下事情。

根據 namespaceId, 和 serviceName 獲取服務信息。
判斷是有有客戶端訂閱了服務，如果由客戶端訂閱了服務，則加入到 UDP 推送列表中，也就是之前我們分析過的 Nacos Server 是如何通知 Nacos Client 服務下線。
閥值判斷，通過各種判斷規(guī)則得到服務列表（判斷詳情請看源碼分析）。
封裝結果集返回。

//com.alibaba.nacos.naming.controllers.InstanceController#doSrvIpxt
public ObjectNode doSrvIpxt(String namespaceId, String serviceName, String agent, String clusters, String clientIP,int udpPort, String env, boolean isCheck, String app, String tid, boolean healthyOnly) throws Exception {//創(chuàng)建客戶端對象ClientInfo clientInfo = new ClientInfo(agent);//創(chuàng)建 ObjectNodeObjectNode result = JacksonUtils.createEmptyJsonNode();//根據  namespaceId serviceName 獲取 serviceService service = serviceManager.getService(namespaceId, serviceName);//緩存時間 默認 10 秒long cacheMillis = switchDomain.getDefaultCacheMillis();// now try to enable the pushtry {// udp 端口大于0 且已經開啟推送  只有客戶端訂閱了  udp  端口才會大于0if (udpPort > 0 && pushService.canEnablePush(agent)) {//添加當前客戶端 IP、UDP端口到 PushService 中 會作為可推送的目標客戶端添加給推送服務組件pushService.addClient(namespaceId, serviceName, clusters, agent, new InetSocketAddress(clientIP, udpPort),pushDataSource, tid, app);//根據服務名 獲取緩存時間默認10 秒cacheMillis = switchDomain.getPushCacheMillis(serviceName);}} catch (Exception e) {Loggers.SRV_LOG.error("[NACOS-API] failed to added push client {}, {}:{}", clientInfo, clientIP, udpPort, e);cacheMillis = switchDomain.getDefaultCacheMillis();}//service 為空判斷if (service == null) {if (Loggers.SRV_LOG.isDebugEnabled()) {Loggers.SRV_LOG.debug("no instance to serve for service: {}", serviceName);}//service 為空 構造空對象 返回result.put("name", serviceName);result.put("clusters", clusters);result.put("cacheMillis", cacheMillis);result.replace("hosts", JacksonUtils.createEmptyArrayNode());return result;}//檢查服務是否禁用 默認是 啟用的checkIfDisabled(service);//服務實例 ipsList<Instance> srvedIPs;//服務實例 ipssrvedIPs = service.srvIPs(Arrays.asList(StringUtils.split(clusters, ",")));// filter ips using selector://若選擇器不空 則根據選擇算法選擇可用的intance列表 默認情況下 選擇器不做任何過濾if (service.getSelector() != null && StringUtils.isNotBlank(clientIP)) {srvedIPs = service.getSelector().select(clientIP, srvedIPs);}//serviceIps 為空判斷if (CollectionUtils.isEmpty(srvedIPs)) {//為空if (Loggers.SRV_LOG.isDebugEnabled()) {Loggers.SRV_LOG.debug("no instance to serve for service: {}", serviceName);}//客戶端類型判斷if (clientInfo.type == ClientInfo.ClientType.JAVA&& clientInfo.version.compareTo(VersionUtil.parseVersion("1.0.0")) >= 0) {result.put("dom", serviceName);} else {result.put("dom", NamingUtils.getServiceName(serviceName));}//構造空對象返回result.put("name", serviceName);result.put("cacheMillis", cacheMillis);result.put("lastRefTime", System.currentTimeMillis());result.put("checksum", service.getChecksum());result.put("useSpecifiedURL", false);result.put("clusters", clusters);result.put("env", env);result.set("hosts", JacksonUtils.createEmptyArrayNode());result.set("metadata", JacksonUtils.transferToJsonNode(service.getMetadata()));return result;}//存儲健康和不健康的實例//key為true的value中存放的是所有健康的instance//key為false的value存放的是所有不健康的instanceMap<Boolean, List<Instance>> ipMap = new HashMap<>(2);ipMap.put(Boolean.TRUE, new ArrayList<>());ipMap.put(Boolean.FALSE, new ArrayList<>());//服務實例遍歷 區(qū)分健康和不健康的實例for (Instance ip : srvedIPs) {ipMap.get(ip.isHealthy()).add(ip);}//閥值判斷 isCheck 客戶端請求中如果沒有傳值 則默認是 fasleif (isCheck) {//false 標識沒有達到保護閥值result.put("reachProtectThreshold", false);}//獲取保護閥值double threshold = service.getProtectThreshold();//通過健康實例除以所有實例 來判斷是否觸發(fā)閥值if ((float) ipMap.get(Boolean.TRUE).size() / srvedIPs.size() <= threshold) {//進入這里標識已經出發(fā)了保護閥值Loggers.SRV_LOG.warn("protect threshold reached, return all ips, service: {}", serviceName);if (isCheck) {//啟動服務保護機制result.put("reachProtectThreshold", true);}//將不健康的實例全部加入到健康的實例中//這樣做的好處是可以保證服務不會那么快被打崩潰 即使有部分失敗的 但是還是有可用的服務 //不健康的實例存在的目的就是分流 緩解健康服務的壓力ipMap.get(Boolean.TRUE).addAll(ipMap.get(Boolean.FALSE));//清空不健康的實例ipMap.get(Boolean.FALSE).clear();}//閥值判斷if (isCheck) {result.put("protectThreshold", service.getProtectThreshold());result.put("reachLocalSiteCallThreshold", false);return JacksonUtils.createEmptyJsonNode();}//能夠走到這里 標識沒有出發(fā) 閥值保護ArrayNode hosts = JacksonUtils.createEmptyArrayNode();//遍歷實例for (Map.Entry<Boolean, List<Instance>> entry : ipMap.entrySet()) {List<Instance> ips = entry.getValue();//如果只需要健康的實例 那就跳過不健康的實例if (healthyOnly && !entry.getKey()) {continue;}//遍歷服務實例for (Instance instance : ips) {// remove disabled instance://移除禁用的實例if (!instance.isEnabled()) {continue;}//創(chuàng)建空對ObjectNode ipObj = JacksonUtils.createEmptyJsonNode();//構建實例對象ipObj.put("ip", instance.getIp());ipObj.put("port", instance.getPort());// deprecated since nacos 1.0.0:ipObj.put("valid", entry.getKey());ipObj.put("healthy", entry.getKey());ipObj.put("marked", instance.isMarked());ipObj.put("instanceId", instance.getInstanceId());ipObj.set("metadata", JacksonUtils.transferToJsonNode(instance.getMetadata()));ipObj.put("enabled", instance.isEnabled());ipObj.put("weight", instance.getWeight());ipObj.put("clusterName", instance.getClusterName());if (clientInfo.type == ClientInfo.ClientType.JAVA&& clientInfo.version.compareTo(VersionUtil.parseVersion("1.0.0")) >= 0) {ipObj.put("serviceName", instance.getServiceName());} else {ipObj.put("serviceName", NamingUtils.getServiceName(instance.getServiceName()));}ipObj.put("ephemeral", instance.isEphemeral());hosts.add(ipObj);}}//設置服務實例列表result.replace("hosts", hosts);//客戶端類型判斷if (clientInfo.type == ClientInfo.ClientType.JAVA&& clientInfo.version.compareTo(VersionUtil.parseVersion("1.0.0")) >= 0) {result.put("dom", serviceName);} else {result.put("dom", NamingUtils.getServiceName(serviceName));}//返回結果result.put("name", serviceName);result.put("cacheMillis", cacheMillis);result.put("lastRefTime", System.currentTimeMillis());result.put("checksum", service.getChecksum());result.put("useSpecifiedURL", false);result.put("clusters", clusters);result.put("env", env);result.replace("metadata", JacksonUtils.transferToJsonNode(service.getMetadata()));return result;
}

PushService#addClient 方法源碼解析

我們上面在分析 InstanceController#doSrvIpxt 方法時候，提到如果客戶端的訂閱了該服務，Nacos 服務端會進行通過 UDP 推送給客戶端最新的服務信息，而這個操作就是由 PushService 類實現的，PushService#addClient 方法只是把服務相關信息加入到了推送列表中。

//com.alibaba.nacos.naming.push.PushService#addClient(java.lang.String, java.lang.String, java.lang.String, java.lang.String, java.net.InetSocketAddress, com.alibaba.nacos.naming.push.DataSource, java.lang.String, java.lang.String)
public void addClient(String namespaceId, String serviceName, String clusters, String agent,InetSocketAddress socketAddr, DataSource dataSource, String tenant, String app) {//構造 PushClient 對象PushClient client = new PushClient(namespaceId, serviceName, clusters, agent, socketAddr, dataSource, tenant,app);//加入到推送列表中addClient(client);
}//com.alibaba.nacos.naming.push.PushService#addClient(com.alibaba.nacos.naming.push.PushService.PushClient)
public void addClient(PushClient client) {// client is stored by key 'serviceName' because notify event is driven by serviceName change//獲取 serviceKeyString serviceKey = UtilsAndCommons.assembleFullServiceName(client.getNamespaceId(), client.getServiceName());//從 客戶端 map  中獲取 當前 client 對象ConcurrentMap<String, PushClient> clients = clientMap.get(serviceKey);//為空 判斷if (clients == null) {//為空 則加入到 客戶端 map 中clientMap.putIfAbsent(serviceKey, new ConcurrentHashMap<>(1024));//加入后獲取clients = clientMap.get(serviceKey);}//獲取之前的 client 對象PushClient oldClient = clients.get(client.toString());if (oldClient != null) {//為空空 刷新 其實是修改 client 最后一次引用時間 可以理解為更新時間oldClient.refresh();} else {//為空 也就是還沒有注冊這個推送目標客戶端  將 client 加入到 clientsPushClient res = clients.putIfAbsent(client.toString(), client);if (res != null) {Loggers.PUSH.warn("client: {} already associated with key {}", res.getAddrStr(), res.toString());}Loggers.PUSH.debug("client: {} added for serviceName: {}", client.getAddrStr(), client.getServiceName());}
}

PushService#onApplicationEvent 方法源碼分析

上面我們分析到客戶端獲取服務信息的時候，服務端會判斷是否有客戶端訂閱了該服務信息，如果有，則會出發(fā)推送給客戶端，最終會把服務信息封裝成一個 PushClient 加入到 clientMap 中，前文我們分析了客戶端是如果感知服務下線的，其中也發(fā)現了一個 clientMap 的存儲結構，而在 PushService#onApplicationEvent 方法會注冊一個延時任務并將該 future 放入 futureMap，該延時任務會從 clientMap 獲取指定namespaceId、 serviceName 的client 集合，遍歷 client 集合，判斷 client 是否是 zombie（僵尸） client，如果是的則移除該 client，否則創(chuàng)建 Receiver.AckEntry，然后通過 UDP 的方式推送給 client，執(zhí)行完畢后會從 futureMap 移除該 future，至此回到了我們前一篇分析的地方，后續(xù)就是我們熟悉的流程，不在重復分析了。

//com.alibaba.nacos.naming.push.PushService#onApplicationEvent
public void onApplicationEvent(ServiceChangeEvent event) {//從事件對象中獲取到 serviceService service = event.getService();//獲取 servicenameString serviceName = service.getName();//獲取名稱空間idString namespaceId = service.getNamespaceId();//使用延時任務 延時1 秒 通過 UDP 的方式來發(fā)送Future future = GlobalExecutor.scheduleUdpSender(() -> {try {Loggers.PUSH.info(serviceName + " is changed, add it to push queue.");//從緩存map中獲取當前服務的內層map 內層map中存放著當前服務的所有Nacos Client的//根據 namespaceId 和 serviceName 獲取對應的 client 信息ConcurrentMap<String, PushClient> clients = clientMap.get(UtilsAndCommons.assembleFullServiceName(namespaceId, serviceName));//為空判斷 如果為空 就沒有必要推送了if (MapUtils.isEmpty(clients)) {return;}//創(chuàng)建緩存 mapMap<String, Object> cache = new HashMap<>(16);//獲取當前時間的 納秒long lastRefTime = System.nanoTime();//遍歷所有的 client 信息for (PushClient client : clients.values()) {//是否是僵尸客戶端if (client.zombie()) {Loggers.PUSH.debug("client is zombie: " + client.toString());//如果是的話 就移除僵尸客戶端clients.remove(client.toString());Loggers.PUSH.debug("client is zombie: " + client.toString());continue;}//ACKReceiver.AckEntry ackEntry;Loggers.PUSH.debug("push serviceName: {} to client: {}", serviceName, client.toString());//獲取推送 keyString key = getPushCacheKey(serviceName, client.getIp(), client.getAgent());byte[] compressData = null;Map<String, Object> data = null;//switchDomain.getDefaultPushCacheMillis() 默認是 10秒 因此不會進入 ifif (switchDomain.getDefaultPushCacheMillis() >= 20000 && cache.containsKey(key)) {org.javatuples.Pair pair = (org.javatuples.Pair) cache.get(key);compressData = (byte[]) (pair.getValue0());data = (Map<String, Object>) pair.getValue1();Loggers.PUSH.debug("[PUSH-CACHE] cache hit: {}:{}", serviceName, client.getAddrStr());}//封裝 ackEntry  將客戶端信息封裝到 ackEntry if (compressData != null) {ackEntry = prepareAckEntry(client, compressData, data, lastRefTime);} else {//這里初始化了需要推送的 客戶端ackEntry = prepareAckEntry(client, prepareHostsData(client), lastRefTime);if (ackEntry != null) {cache.put(key, new org.javatuples.Pair<>(ackEntry.origin.getData(), ackEntry.data));}}Loggers.PUSH.info("serviceName: {} changed, schedule push for: {}, agent: {}, key: {}",client.getServiceName(), client.getAddrStr(), client.getAgent(),(ackEntry == null ? null : ackEntry.key));//通過 udp 協議向 Nacos 客戶端推送數據udpPush(ackEntry);}} catch (Exception e) {Loggers.PUSH.error("[NACOS-PUSH] failed to push serviceName: {} to client, error: {}", serviceName, e);} finally {//移除 futurefutureMap.remove(UtilsAndCommons.assembleFullServiceName(namespaceId, serviceName));}}, 1000, TimeUnit.MILLISECONDS);//任務放入 futureMap  表示已經發(fā)送了 udp 到客戶端的服務實例futureMap.put(UtilsAndCommons.assembleFullServiceName(namespaceId, serviceName), future);}