繼 Nacos服務注冊原理 后，我們來看一下Nacos 是怎么實現服務的消費的。
服務注冊成功之后，消費者就可以從nacos server中獲取到服務提供者的地址，然后進行服務的調用。在服務消費中，有一個核心的類 NacosDiscoveryClient 來負責和nacos交互，去獲得服務提供者的地址信息。基於org.springframework.cloud.client.discovery.DiscoveryClient 的實現，如下圖所示，Consul、Eureka是我們所熟悉的。他們所實現的是同一套規范。
NacosDiscoveryClient 中提供了一個 getInstances 方法用來根據服務提供者名稱獲取服務提供者的url地址的方法.
客戶端啟動獲取服務列表：
我們可以通過Debug 模式來驗證這一猜想，啟動服務消費者一定會進入NacosDiscoveryClient 的 getInstances 方法。
@Override public List<ServiceInstance> getInstances(String serviceId) { try { return serviceDiscovery.getInstances(serviceId); } catch (Exception e) { throw new RuntimeException( "Can not get hosts from nacos server. serviceId: " + serviceId, e); } }
然后回調用 NacosServiceDiscovery 的 getInstances 方法，講我們所配置的 group 、serviceId 傳過去，獲取基於該serviceId的實例列表。
調用NamingService，根據serviceId、group獲得服務實例列表。然后把instance轉化為ServiceInstance對象
public List<ServiceInstance> getInstances(String serviceId) throws NacosException { String group = discoveryProperties.getGroup(); List<Instance> instances = discoveryProperties.namingServiceInstance() .selectInstances(serviceId, group, true); return hostToServiceInstanceList(instances, serviceId); } @ConfigurationProperties("spring.cloud.nacos.discovery") public class NacosDiscoveryProperties { //...
}
NacosNamingService.selectInstances 首先從 hostReactor 獲取 serviceInfo，然后再從serviceInfo.getHosts()剔除非 healty、非enabled、weight小於等於0的 instance 再返回；如果subscribe為true，則執行 hostReactor.getServiceInfo獲取serviceInfo，否則執行
hostReactor.getServiceInfoDirectlyFromServer獲取serviceInfo
@Override public List<Instance> selectInstances(String serviceName, String groupName, List<String> clusters, boolean healthy, boolean subscribe) throws NacosException { ServiceInfo serviceInfo; if (subscribe) {//是否訂閱服務地址的變化，默認為true
serviceInfo = hostReactor.getServiceInfo(NamingUtils.getGroupedName(serviceName, groupName), StringUtils.join(clusters, ",")); } else { serviceInfo = hostReactor.getServiceInfoDirectlyFromServer(NamingUtils.getGroupedName(serviceName, groupName), StringUtils.join(clusters, ",")); } return selectInstances(serviceInfo, healthy); } private List<Instance> selectInstances(ServiceInfo serviceInfo, boolean healthy) { List<Instance> list; if (serviceInfo == null || CollectionUtils.isEmpty(list = serviceInfo.getHosts())) { return new ArrayList<Instance>(); } Iterator<Instance> iterator = list.iterator(); while (iterator.hasNext()) { Instance instance = iterator.next(); if (healthy != instance.isHealthy() || !instance.isEnabled() || instance.getWeight() <= 0) { iterator.remove(); } } return list; }
從 hostReactor 獲取 serviceInfo的具體操作如下：
public ServiceInfo getServiceInfo(final String serviceName, final String clusters) { NAMING_LOGGER.debug("failover-mode: " + failoverReactor.isFailoverSwitch()); //拼接服務名稱+集群名稱（默認為空）
String key = ServiceInfo.getKey(serviceName, clusters); if (failoverReactor.isFailoverSwitch()) { return failoverReactor.getService(key); } //從ServiceInfoMap中根據key來查找服務提供者列表，ServiceInfoMap是客戶端的服務地址的本地緩存
ServiceInfo serviceObj = getSerivceInfo0(serviceName, clusters); //如果為空，表示本地緩存不存在
if (null == serviceObj) { serviceObj = new ServiceInfo(serviceName, clusters); //如果找不到則創建一個新的然后放入serviceInfoMap，同時放入updatingMap，執行updateServiceNow，再從updatingMap移除；
serviceInfoMap.put(serviceObj.getKey(), serviceObj); updatingMap.put(serviceName, new Object()); // 立馬從Nacos server中去加載服務地址信息
updateServiceNow(serviceName, clusters); updatingMap.remove(serviceName); } else if (updatingMap.containsKey(serviceName)) { //如果從serviceInfoMap找出來的serviceObj在updatingMap中則等待UPDATE_HOLD_INTERVAL
if (updateHoldInterval > 0) { // hold a moment waiting for update finish
synchronized (serviceObj) { try { serviceObj.wait(updateHoldInterval); } catch (InterruptedException e) { NAMING_LOGGER.error("[getServiceInfo] serviceName:" + serviceName + ", clusters:" + clusters, e); } } } } // 開啟定時調度，每10s去查詢一次服務地址 //如果本地緩存中存在，則通過scheduleUpdateIfAbsent開啟定時任務，再從serviceInfoMap取出serviceInfo
scheduleUpdateIfAbsent(serviceName, clusters); return serviceInfoMap.get(serviceObj.getKey()); }
其中獲取服務實例列表信息的方法為 updateServiceNow
public void updateServiceNow(String serviceName, String clusters) { ServiceInfo oldService = getSerivceInfo0(serviceName, clusters); try { String result = serverProxy.queryList(serviceName, clusters, pushReceiver.getUDPPort(), false); if (StringUtils.isNotEmpty(result)) { processServiceJSON(result); } // ....... }
可以發現這里請求列表的時候發送了一個 pushReceiver.getUDPPort() ,這就是我們在服務注冊的時候提到的，Nacos Server在檢測到心跳超時的時候回主動發起一下UDP請求向客戶端發送服務注冊信息。哪個UDP端口就是這里傳輸給NacosServer的。
可以看到 queryList
public String queryList(String serviceName, String clusters, int udpPort, boolean healthyOnly) throws NacosException { // 組裝請求參數
final Map<String, String> params = new HashMap<String, String>(8); params.put(CommonParams.NAMESPACE_ID, namespaceId); params.put(CommonParams.SERVICE_NAME, serviceName); params.put("clusters", clusters); params.put("udpPort", String.valueOf(udpPort)); params.put("clientIP", NetUtils.localIP()); params.put("healthyOnly", String.valueOf(healthyOnly)); //通過HttpClient 發送請求
return reqAPI(UtilAndComs.NACOS_URL_BASE + "/instance/list", params, HttpMethod.GET); }
Nacos Server 處理消費端請求：
通過上面消費端的請求 URL，我們可以定位到服務端源碼的 InstanceController 的對應 GET請求的列表獲取接口：
@GetMapping("/list") @Secured(parser = NamingResourceParser.class, action = ActionTypes.READ) public ObjectNode list(HttpServletRequest request) throws Exception { String namespaceId = WebUtils.optional(request, CommonParams.NAMESPACE_ID, Constants.DEFAULT_NAMESPACE_ID); //從 request中獲取請求參數
String serviceName = WebUtils.required(request, CommonParams.SERVICE_NAME); String agent = WebUtils.getUserAgent(request); String clusters = WebUtils.optional(request, "clusters", StringUtils.EMPTY); String clientIP = WebUtils.optional(request, "clientIP", StringUtils.EMPTY); int udpPort = Integer.parseInt(WebUtils.optional(request, "udpPort", "0")); 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); }
就跟查詢數據庫一樣，現在有參數了，接下去就是重頭戲了：
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 { // 創建一個客戶端的信息 ClientInfo clientInfo = new ClientInfo(agent);
// 准備返回結果類型 ObjectNode result = JacksonUtils.createEmptyJsonNode();
// 從緩存的 serviceMap中獲取相應的服務實例 Service service = serviceManager.getService(namespaceId, serviceName); long cacheMillis = switchDomain.getDefaultCacheMillis(); // now try to enable the push
try {//這里判斷udp端口跟是否開啟推送機制 if (udpPort > 0 && pushService.canEnablePush(agent)) { //這里就很熟悉了，將構建一個InetSocketAddress，將Nacos Server 作為客戶端，請求消費端進行推送 pushService .addClient(namespaceId, serviceName, clusters, agent, new InetSocketAddress(clientIP, udpPort), pushDataSource, tid, app); 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(); } if (service == null) {//如果獲取到的服務為空，組裝結果返回 if (Loggers.SRV_LOG.isDebugEnabled()) { Loggers.SRV_LOG.debug("no instance to serve for service: {}", serviceName); } result.put("name", serviceName); result.put("clusters", clusters); result.put("cacheMillis", cacheMillis);
// 返回空的 hosts result.replace("hosts", JacksonUtils.createEmptyArrayNode()); return result; } //檢查服務是否可用 checkIfDisabled(service); //准備返回的實例列表 List<Instance> srvedIPs; // 通過傳進來的 clusters 獲取服務ips srvedIPs = service.srvIPs(Arrays.asList(StringUtils.split(clusters, ","))); // filter ips using selector:
if (service.getSelector() != null && StringUtils.isNotBlank(clientIP)) { srvedIPs = service.getSelector().select(clientIP, srvedIPs); } //很顯然，這里獲取的sevedIPs為空，因為我們clusters是空的 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; } //兩個List是分別放置健康/非健康實例 Map<Boolean, List<Instance>> ipMap = new HashMap<>(2); ipMap.put(Boolean.TRUE, new ArrayList<>()); ipMap.put(Boolean.FALSE, new ArrayList<>()); //篩選健康實例 for (Instance ip : srvedIPs) { ipMap.get(ip.isHealthy()).add(ip); } if (isCheck) { result.put("reachProtectThreshold", false); } //這個類似於Eureka的自我保護機制。避免網絡延遲帶來的心跳超時的實例剔除 double threshold = service.getProtectThreshold(); if ((float) ipMap.get(Boolean.TRUE).size() / srvedIPs.size() <= threshold) { 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(); } ArrayNode hosts = JacksonUtils.createEmptyArrayNode(); // 遍歷map，組裝數據返回給消費者 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; } 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; }
經過這么一系列操作以后，服務消費者就能獲取到相應的服務實例集合了。
服務動態更新：
基於上面的分析，服務消費者對於服務實例的動態更新主要來源於兩個地方，第一個就是本地的定時任務，第二個就是采用服務端的 Push 機制，如下圖。
pull 定時任務請求更新服務信息：
在查詢服務調用 getServiceInfo 方法的代碼中，會開啟一個定時任務，這個任務會在默認在1s之后開始執行。而任務的具體實現是一個UpdateTask。
public void scheduleUpdateIfAbsent(String serviceName, String clusters) { if (futureMap.get(ServiceInfo.getKey(serviceName, clusters)) != null) { return; } synchronized (futureMap) { if (futureMap.get(ServiceInfo.getKey(serviceName, clusters)) != null) { return; } ScheduledFuture<?> future = addTask(new UpdateTask(serviceName, clusters)); futureMap.put(ServiceInfo.getKey(serviceName, clusters), future); } }
所以我們定位到 UpdateTask 的 run 方法：
@Override public void run() { try {//查詢本地緩存
ServiceInfo serviceObj = serviceInfoMap.get(ServiceInfo.getKey(serviceName, clusters)); //如果本地緩存為空，則向服務器發起更新請求
if (serviceObj == null) { updateServiceNow(serviceName, clusters); // 開啟一個任務，延后一秒執行一次
executor.schedule(this, DEFAULT_DELAY, TimeUnit.MILLISECONDS); return; } //判斷服務是否已過期
if (serviceObj.getLastRefTime() <= lastRefTime) { updateServiceNow(serviceName, clusters); serviceObj = serviceInfoMap.get(ServiceInfo.getKey(serviceName, clusters)); } else { // if serviceName already updated by push, we should not override it // since the push data may be different from pull through force push //如果服務已經被基於push機制的情況下做了更新，那么我們不需要覆蓋本地服務。 //因為push過來的數據和pull數據不同，所以這里只是調用請求去刷新服務
refreshOnly(serviceName, clusters); } //延后10s執行
executor.schedule(this, serviceObj.getCacheMillis(), TimeUnit.MILLISECONDS); //更新最后一次刷新時間
lastRefTime = serviceObj.getLastRefTime(); } catch (Throwable e) { NAMING_LOGGER.warn("[NA] failed to update serviceName: " + serviceName, e); } }
push請求推送數據:
還記得在服務提供者發起服務注冊時。在 createEmptyService 方法中，會創建一個空的服務.並且在這個創建過程中，調用了一個 putServiceAndInit ，這個方法中除了創建空的服務並且初始化，還會調用 service.init 方法進行服務的初始化。
private void putServiceAndInit(Service service) throws NacosException { putService(service); service.init(); consistencyService .listen(KeyBuilder.buildInstanceListKey(service.getNamespaceId(), service.getName(), true), service); consistencyService .listen(KeyBuilder.buildInstanceListKey(service.getNamespaceId(), service.getName(), false), service); Loggers.SRV_LOG.info("[NEW-SERVICE] {}", service.toJson()); } /** * Init service. */
public void init() { HealthCheckReactor.scheduleCheck(clientBeatCheckTask); for (Map.Entry<String, Cluster> entry : clusterMap.entrySet()) { entry.getValue().setService(this); entry.getValue().init(); } }
這個init方法，會和當前服務提供者建立一個心跳檢測機制，這個心跳檢測會每5s執行一次。然后來看 ClientBeatCheckTask.run
@Override public void run() { try { if (!getDistroMapper().responsible(service.getName())) { return; } if (!getSwitchDomain().isHealthCheckEnabled()) { return; } //獲取到所有服務實例
List<Instance> instances = service.allIPs(true); // first set health status of instances: //遍歷服務節點進行心跳檢測
for (Instance instance : instances) { //如果服務實例的最后一次心跳時間大於設置的超時時間，則認為這個服務已經下線。
if (System.currentTimeMillis() - instance.getLastBeat() > instance.getInstanceHeartBeatTimeOut()) { if (!instance.isMarked()) { if (instance.isHealthy()) { instance.setHealthy(false); Loggers.EVT_LOG .info("{POS} {IP-DISABLED} valid: {}:{}@{}@{}, region: {}, msg: client timeout after {}, last beat: {}", instance.getIp(), instance.getPort(), instance.getClusterName(), service.getName(), UtilsAndCommons.LOCALHOST_SITE, instance.getInstanceHeartBeatTimeOut(), instance.getLastBeat()); getPushService().serviceChanged(service);//推送服務變更事 //發布實例心跳超時事件
ApplicationUtils.publishEvent(new InstanceHeartbeatTimeoutEvent(this, instance)); } } } } if (!getGlobalConfig().isExpireInstance()) { return; } // then remove obsolete instances:
for (Instance instance : instances) { if (instance.isMarked()) { continue; } if (System.currentTimeMillis() - instance.getLastBeat() > instance.getIpDeleteTimeout()) { // delete instance
Loggers.SRV_LOG.info("[AUTO-DELETE-IP] service: {}, ip: {}", service.getName(), JacksonUtils.toJson(instance)); deleteIp(instance);//刪除過期的服務實例
} } } catch (Exception e) { Loggers.SRV_LOG.warn("Exception while processing client beat time out.", e); } }
在這里 getPushService().serviceChanged(service) 會發布一個服務變更事件：
public void serviceChanged(Service service) { // merge some change events to reduce the push frequency:
if (futureMap .containsKey(UtilsAndCommons.assembleFullServiceName(service.getNamespaceId(), service.getName()))) { return; } this.applicationContext.publishEvent(new ServiceChangeEvent(this, service)); }
而 PushService 類實現了 ApplicationListener<ServiceChangeEvent> 所以本身又會取監聽該事件，監聽服務狀態變更事件，然后遍歷所有的客戶端，通過udp協議進行消息的廣播通知：
@Override public void onApplicationEvent(ServiceChangeEvent event) { Service service = event.getService();//獲取到服務
String serviceName = service.getName();//服務名
String namespaceId = service.getNamespaceId();//命名空間 //執行任務
Future future = GlobalExecutor.scheduleUdpSender(() -> { try { Loggers.PUSH.info(serviceName + " is changed, add it to push queue."); ConcurrentMap<String, PushClient> clients = clientMap .get(UtilsAndCommons.assembleFullServiceName(namespaceId, serviceName)); if (MapUtils.isEmpty(clients)) { return; } Map<String, Object> cache = new HashMap<>(16); long lastRefTime = System.nanoTime(); 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; } Receiver.AckEntry ackEntry; Loggers.PUSH.debug("push serviceName: {} to client: {}", serviceName, client.toString()); String key = getPushCacheKey(serviceName, client.getIp(), client.getAgent()); byte[] compressData = null; Map<String, Object> data = null; if (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()); } 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 推送
udpPush(ackEntry); } } catch (Exception e) { Loggers.PUSH.error("[NACOS-PUSH] failed to push serviceName: {} to client, error: {}", serviceName, e); } finally { futureMap.remove(UtilsAndCommons.assembleFullServiceName(namespaceId, serviceName)); } }, 1000, TimeUnit.MILLISECONDS); futureMap.put(UtilsAndCommons.assembleFullServiceName(namespaceId, serviceName), future); }
那么服務消費者此時應該是建立了一個udp服務的監聽，否則服務端無法進行數據的推送。這個監聽是在HostReactor的構造方法中初始化的
public HostReactor(EventDispatcher eventDispatcher, NamingProxy serverProxy, String cacheDir, boolean loadCacheAtStart, int pollingThreadCount) { executor = new ScheduledThreadPoolExecutor(pollingThreadCount, new ThreadFactory() { @Override public Thread newThread(Runnable r) { Thread thread = new Thread(r); thread.setDaemon(true); thread.setName("com.alibaba.nacos.client.naming.updater"); return thread; } }); this.eventDispatcher = eventDispatcher; this.serverProxy = serverProxy; this.cacheDir = cacheDir; if (loadCacheAtStart) { this.serviceInfoMap = new ConcurrentHashMap<String, ServiceInfo>(DiskCache.read(this.cacheDir)); } else { this.serviceInfoMap = new ConcurrentHashMap<String, ServiceInfo>(16); } this.updatingMap = new ConcurrentHashMap<String, Object>(); this.failoverReactor = new FailoverReactor(this, cacheDir); this.pushReceiver = new PushReceiver(this); }
這里主要看 new PushReceiver(this) 把this 傳進去，初始化了一個DatagramSocket，這是一個Udp的socket連接，開啟一個線程，定時執行當前任務
public PushReceiver(HostReactor hostReactor) { try { this.hostReactor = hostReactor; udpSocket = new DatagramSocket(); executorService = new ScheduledThreadPoolExecutor(1, new ThreadFactory() { @Override public Thread newThread(Runnable r) { Thread thread = new Thread(r); thread.setDaemon(true); thread.setName("com.alibaba.nacos.naming.push.receiver"); return thread; } }); executorService.execute(this); } catch (Exception e) { NAMING_LOGGER.error("[NA] init udp socket failed", e); } }
然后需要關注的是 PushReceiver 的 Run 方法：在run方法中，不斷循環監聽服務端的push請求。然后調用 processServiceJSON 對服務端的數據進行解析。
@Override public void run() { while (true) { try { // byte[] is initialized with 0 full filled by default
byte[] buffer = new byte[UDP_MSS]; DatagramPacket packet = new DatagramPacket(buffer, buffer.length); udpSocket.receive(packet); String json = new String(IoUtils.tryDecompress(packet.getData()), "UTF-8").trim(); NAMING_LOGGER.info("received push data: " + json + " from " + packet.getAddress().toString()); PushPacket pushPacket = JSON.parseObject(json, PushPacket.class); String ack; if ("dom".equals(pushPacket.type) || "service".equals(pushPacket.type)) { hostReactor.processServiceJSON(pushPacket.data); // send ack to server
ack = "{\"type\": \"push-ack\""
+ ", \"lastRefTime\":\"" + pushPacket.lastRefTime + "\", \"data\":" + "\"\"}"; } else if ("dump".equals(pushPacket.type)) { // dump data to server
ack = "{\"type\": \"dump-ack\""
+ ", \"lastRefTime\": \"" + pushPacket.lastRefTime + "\", \"data\":" + "\""
+ StringUtils.escapeJavaScript(JSON.toJSONString(hostReactor.getServiceInfoMap())) + "\"}"; } else { // do nothing send ack only
ack = "{\"type\": \"unknown-ack\""
+ ", \"lastRefTime\":\"" + pushPacket.lastRefTime + "\", \"data\":" + "\"\"}"; } udpSocket.send(new DatagramPacket(ack.getBytes(Charset.forName("UTF-8")), ack.getBytes(Charset.forName("UTF-8")).length, packet.getSocketAddress())); } catch (Exception e) { NAMING_LOGGER.error("[NA] error while receiving push data", e); } } }
就這樣完成服務的動態更新。更加細節的部分請閱讀源碼實現。