updateIPs 方法会将服务实例信息，更新到注册表的内存中去，并且会以 udp 的方式通知当前服务的订阅者。
public void updateIPs(Collection<Instance> instances, boolean ephemeral) { Map<String, List<Instance>> ipMap = new HashMap<>(clusterMap.size()); for (String clusterName : clusterMap.keySet()) { ipMap.put(clusterName, new ArrayList<>()); } for (Instance instance : instances) { try { if (instance == null) { Loggers.SRV_LOG.error("[NACOS-DOM] received malformed ip: null"); continue; } if (StringUtils.isEmpty(instance.getClusterName())) { instance.setClusterName(UtilsAndCommons.DEFAULT_CLUSTER_NAME); } if (!clusterMap.containsKey(instance.getClusterName())) { Loggers.SRV_LOG .warn("cluster: {} not found, ip: {}, will create new cluster with default configuration.", instance.getClusterName(), instance.toJson()); Cluster cluster = new Cluster(instance.getClusterName(), this); cluster.init(); getClusterMap().put(instance.getClusterName(), cluster); } List<Instance> clusterIPs = ipMap.get(instance.getClusterName()); if (clusterIPs == null) { clusterIPs = new LinkedList<>(); ipMap.put(instance.getClusterName(), clusterIPs); } clusterIPs.add(instance); } catch (Exception e) { Loggers.SRV_LOG.error("[NACOS-DOM] failed to process ip: " + instance, e); } } for (Map.Entry<String, List<Instance>> entry : ipMap.entrySet()) { //make every ip mine List<Instance> entryIPs = entry.getValue(); // 更新服务列表 clusterMap.get(entry.getKey()).updateIps(entryIPs, ephemeral); } setLastModifiedMillis(System.currentTimeMillis()); // 推送服务订阅者消息 getPushService().serviceChanged(this); StringBuilder stringBuilder = new StringBuilder(); for (Instance instance : allIPs()) { stringBuilder.append(instance.toIpAddr()).append("_").append(instance.isHealthy()).append(","); } Loggers.EVT_LOG.info("[IP-UPDATED] namespace: {}, service: {}, ips: {}", getNamespaceId(), getName(), stringBuilder.toString()); }
CP 模式
Nacos 默认就是采用的 CP 模式使用 Raft 协议实现。实现类是 PersistentConsistencyServiceDelegateImpl
首先我们先看他的 put 方法
public void put(String key, Record value) throws NacosException { checkIsStopWork(); try { raftCore.signalPublish(key, value); } catch (Exception e) { Loggers.RAFT.error("Raft put failed.", e); throw new NacosException(NacosException.SERVER_ERROR, "Raft put failed, key:" + key + ", value:" + value, e); } }
调用 raftCore.signalPublish(key, value); 主要的步骤如下
- 判断是否是 Leader 节点，如果不是 Leader 节点将请求转发给 Leader 节点处理；
- 如果是 Leader 节点，首先执行
onPublish(datum, peers.local());方法，内部首先通过raftStore.updateTerm(local.term.get());方法持久化到文件，然后通过NotifyCenter.publishEvent(ValueChangeEvent.builder().key(datum.key).action(DataOperation.CHANGE).build());异步更新到内存；
- 通过 CountDownLatch 实现了一个过半机制
new CountDownLatch(peers.majorityCount())只有当成功的节点大于 N/2 + 1 的时候才返回成功。
- 调用其他的 Nacos 节点的
/raft/datum/commit同步实例信息。
public void signalPublish(String key, Record value) throws Exception { if (stopWork) { throw new IllegalStateException("old raft protocol already stop work"); } if (!isLeader()) { ObjectNode params = JacksonUtils.createEmptyJsonNode(); params.put("key", key); params.replace("value", JacksonUtils.transferToJsonNode(value)); Map<String, String> parameters = new HashMap<>(1); parameters.put("key", key); final RaftPeer leader = getLeader(); raftProxy.proxyPostLarge(leader.ip, API_PUB, params.toString(), parameters); return; } OPERATE_LOCK.lock(); try { final long start = System.currentTimeMillis(); final Datum datum = new Datum(); datum.key = key; datum.value = value; if (getDatum(key) == null) { datum.timestamp.set(1L); } else { datum.timestamp.set(getDatum(key).timestamp.incrementAndGet()); } ObjectNode json = JacksonUtils.createEmptyJsonNode(); json.replace("datum", JacksonUtils.transferToJsonNode(datum)); json.replace("source", JacksonUtils.transferToJsonNode(peers.local())); onPublish(datum, peers.local()); final String content = json.toString(); final CountDownLatch latch = new CountDownLatch(peers.majorityCount()); for (final String server : peers.allServersIncludeMyself()) { if (isLeader(server)) { latch.countDown(); continue; } final String url = buildUrl(server, API_ON_PUB); HttpClient.asyncHttpPostLarge(url, Arrays.asList("key", key), content, new Callback<String>() { @Override public void onReceive(RestResult<String> result) { if (!result.ok()) { Loggers.RAFT .warn("[RAFT] failed to publish data to peer, datumId={}, peer={}, http code={}", datum.key, server, result.getCode()); return; } latch.countDown(); } @Override public void onError(Throwable throwable) { Loggers.RAFT.error("[RAFT] failed to publish data to peer", throwable); } @Override public void onCancel() { } }); } if (!latch.await(UtilsAndCommons.RAFT_PUBLISH_TIMEOUT, TimeUnit.MILLISECONDS)) { // only majority servers return success can we consider this update success Loggers.RAFT.error("data publish failed, caused failed to notify majority, key={}", key); throw new IllegalStateException("data publish failed, caused failed to notify majority, key=" + key); } long end = System.currentTimeMillis(); Loggers.RAFT.info("signalPublish cost {} ms, key: {}", (end - start), key); } finally { OPERATE_LOCK.unlock(); } }
判断 AP 模式还是 CP 模式
如果注册 nacos 的 client 节点注册时 ephemeral=true，那么 nacos 集群对这个 client 节点的效果就是 ap 的采用 distro，而注册nacos 的 client 节点注册时 ephemeral=false，那么nacos 集群对这个节点的效果就是 cp 的采用 raft。根据 client 注册时的属性，ap，cp 同时混合存在，只是对不同的 client 节点效果不同
Nacos 源码调试
Nacos 启动文件
首先我们需要找到 Nacos 的启动类，首先需要找到启动的 jar.
然后我们在解压 target/nacos-server.jar
解压命令：
# 解压 jar 包 tar -zxvf nacos-server.jar # 查看 MANIFEST.MF 内容 cat META-INF/MANIFEST.MF Manifest-Version: 1.0 Implementation-Title: nacos-console 1.4.2 Implementation-Version: 1.4.2 Archiver-Version: Plexus Archiver Built-By: xiweng.yy Spring-Boot-Layers-Index: BOOT-INF/layers.idx Specification-Vendor: Alibaba Group Specification-Title: nacos-console 1.4.2 Implementation-Vendor-Id: com.alibaba.nacos Spring-Boot-Version: 2.5.0-RC1 Implementation-Vendor: Alibaba Group Main-Class: org.springframework.boot.loader.PropertiesLauncher Spring-Boot-Classpath-Index: BOOT-INF/classpath.idx Start-Class: com.alibaba.nacos.Nacos Spring-Boot-Classes: BOOT-INF/classes/ Spring-Boot-Lib: BOOT-INF/lib/ Created-By: Apache Maven 3.6.3 Build-Jdk: 1.8.0_231 Specification-Version: 1.4.2
通过 MANIFEST.MF 中的配置信息，我们可以找到 Start-Class 这个配置这个类就是 Spring-Boot 项目的启动类 com.alibaba.nacos.Nacos
Nacos 调试
通过 com.alibaba.nacos.Nacos 的启动类，我们可以通过这个类在 Idea 中进行启动，然后调试。
