扫二维码与项目经理沟通
我们在微信上24小时期待你的声音
解答本文疑问/技术咨询/运营咨询/技术建议/互联网交流
本篇内容主要讲解“nacos ServiceManager中UpdatedServiceProcessor的原理和作用是什么”,感兴趣的朋友不妨来看看。本文介绍的方法操作简单快捷,实用性强。下面就让小编来带大家学习“nacos ServiceManager中UpdatedServiceProcessor的原理和作用是什么”吧!
宾县网站建设公司创新互联,宾县网站设计制作,有大型网站制作公司丰富经验。已为宾县数千家提供企业网站建设服务。企业网站搭建\成都外贸网站制作要多少钱,请找那个售后服务好的宾县做网站的公司定做!
本文主要研究一下nacos ServiceManager的UpdatedServiceProcessor
nacos-1.1.3/naming/src/main/java/com/alibaba/nacos/naming/core/ServiceManager.java
@Component @DependsOn("nacosApplicationContext") public class ServiceManager implements RecordListener{ /** * Map > */ private Map > serviceMap = new ConcurrentHashMap<>(); private LinkedBlockingDeque toBeUpdatedServicesQueue = new LinkedBlockingDeque<>(1024 * 1024); private Synchronizer synchronizer = new ServiceStatusSynchronizer(); private final Lock lock = new ReentrantLock(); @Resource(name = "consistencyDelegate") private ConsistencyService consistencyService; @Autowired private SwitchDomain switchDomain; @Autowired private DistroMapper distroMapper; @Autowired private ServerListManager serverListManager; @Autowired private PushService pushService; private final Object putServiceLock = new Object(); @PostConstruct public void init() { UtilsAndCommons.SERVICE_SYNCHRONIZATION_EXECUTOR.schedule(new ServiceReporter(), 60000, TimeUnit.MILLISECONDS); UtilsAndCommons.SERVICE_UPDATE_EXECUTOR.submit(new UpdatedServiceProcessor()); try { Loggers.SRV_LOG.info("listen for service meta change"); consistencyService.listen(KeyBuilder.SERVICE_META_KEY_PREFIX, this); } catch (NacosException e) { Loggers.SRV_LOG.error("listen for service meta change failed!"); } } //...... }
ServiceManager的init方法往UtilsAndCommons.SERVICE_UPDATE_EXECUTOR提交了UpdatedServiceProcessor任务
nacos-1.1.3/naming/src/main/java/com/alibaba/nacos/naming/core/ServiceManager.java
private class UpdatedServiceProcessor implements Runnable { //get changed service from other server asynchronously @Override public void run() { ServiceKey serviceKey = null; try { while (true) { try { serviceKey = toBeUpdatedServicesQueue.take(); } catch (Exception e) { Loggers.EVT_LOG.error("[UPDATE-DOMAIN] Exception while taking item from LinkedBlockingDeque."); } if (serviceKey == null) { continue; } GlobalExecutor.submitServiceUpdate(new ServiceUpdater(serviceKey)); } } catch (Exception e) { Loggers.EVT_LOG.error("[UPDATE-DOMAIN] Exception while update service: {}", serviceKey, e); } } }
UpdatedServiceProcessor实现了Runnable方法,其run方法会不断循环从toBeUpdatedServicesQueue获取元素,然后使用GlobalExecutor.submitServiceUpdate提交ServiceUpdater
nacos-1.1.3/naming/src/main/java/com/alibaba/nacos/naming/core/ServiceManager.java
private class ServiceUpdater implements Runnable { String namespaceId; String serviceName; String serverIP; public ServiceUpdater(ServiceKey serviceKey) { this.namespaceId = serviceKey.getNamespaceId(); this.serviceName = serviceKey.getServiceName(); this.serverIP = serviceKey.getServerIP(); } @Override public void run() { try { updatedHealthStatus(namespaceId, serviceName, serverIP); } catch (Exception e) { Loggers.SRV_LOG.warn("[DOMAIN-UPDATER] Exception while update service: {} from {}, error: {}", serviceName, serverIP, e); } } }
ServiceUpdater实现了Runnable接口,其run方法执行的是updatedHealthStatus
nacos-1.1.3/naming/src/main/java/com/alibaba/nacos/naming/core/ServiceManager.java
@Component @DependsOn("nacosApplicationContext") public class ServiceManager implements RecordListener{ /** * Map > */ private Map > serviceMap = new ConcurrentHashMap<>(); private LinkedBlockingDeque toBeUpdatedServicesQueue = new LinkedBlockingDeque<>(1024 * 1024); private Synchronizer synchronizer = new ServiceStatusSynchronizer(); private final Lock lock = new ReentrantLock(); @Resource(name = "consistencyDelegate") private ConsistencyService consistencyService; @Autowired private SwitchDomain switchDomain; @Autowired private DistroMapper distroMapper; @Autowired private ServerListManager serverListManager; @Autowired private PushService pushService; private final Object putServiceLock = new Object(); //...... public void updatedHealthStatus(String namespaceId, String serviceName, String serverIP) { Message msg = synchronizer.get(serverIP, UtilsAndCommons.assembleFullServiceName(namespaceId, serviceName)); JSONObject serviceJson = JSON.parseObject(msg.getData()); JSONArray ipList = serviceJson.getJSONArray("ips"); Map ipsMap = new HashMap<>(ipList.size()); for (int i = 0; i < ipList.size(); i++) { String ip = ipList.getString(i); String[] strings = ip.split("_"); ipsMap.put(strings[0], strings[1]); } Service service = getService(namespaceId, serviceName); if (service == null) { return; } boolean changed = false; List instances = service.allIPs(); for (Instance instance : instances) { boolean valid = Boolean.parseBoolean(ipsMap.get(instance.toIPAddr())); if (valid != instance.isHealthy()) { changed = true; instance.setHealthy(valid); Loggers.EVT_LOG.info("{} {SYNC} IP-{} : {}@{}{}", serviceName, (instance.isHealthy() ? "ENABLED" : "DISABLED"), instance.getIp(), instance.getPort(), instance.getClusterName()); } } if (changed) { pushService.serviceChanged(service); } StringBuilder stringBuilder = new StringBuilder(); List allIps = service.allIPs(); for (Instance instance : allIps) { stringBuilder.append(instance.toIPAddr()).append("_").append(instance.isHealthy()).append(","); } if (changed && Loggers.EVT_LOG.isDebugEnabled()) { Loggers.EVT_LOG.debug("[HEALTH-STATUS-UPDATED] namespace: {}, service: {}, ips: {}", service.getNamespaceId(), service.getName(), stringBuilder.toString()); } } //...... }
updatedHealthStatus方法会从synchronizer获取msg,组装ipsMap,之后通过service.allIPs()获取instances信息,然后遍历instances从ipsMap获取实例的valid状态,如果与instance的isHealthy()对不上则标记为changed,更新instance的healthy;对于changed的则通过pushService.serviceChanged(service)发布事件,最后打印日志
ServiceManager的init方法往UtilsAndCommons.SERVICE_UPDATE_EXECUTOR提交了UpdatedServiceProcessor任务
UpdatedServiceProcessor实现了Runnable方法,其run方法会不断循环从toBeUpdatedServicesQueue获取元素,然后使用GlobalExecutor.submitServiceUpdate提交ServiceUpdater
ServiceUpdater实现了Runnable接口,其run方法执行的是updatedHealthStatus
到此,相信大家对“nacos ServiceManager中UpdatedServiceProcessor的原理和作用是什么”有了更深的了解,不妨来实际操作一番吧!这里是创新互联网站,更多相关内容可以进入相关频道进行查询,关注我们,继续学习!
我们在微信上24小时期待你的声音
解答本文疑问/技术咨询/运营咨询/技术建议/互联网交流