nacos1.4-CP架构源码

本文主要介绍nacos1.4的CP架构，nacos通过raft协议（半数以上成功）来控制集群的强一致性，在源代码中使用到countdownlatch锁来控制半数以上成功。

1.Raft协议

演示网址：http://thesecretlivesofdata.com/raft/

分区容错性：针对多节点的系统，分区指网络分区（由于网络原因节点之间无法通信同步数据），容错指系统节点出现出现分去了，对外依然要提供服务，不能因为分区而导致整个系统不能提供服务。

在集群状态下，CP保证半数以上写入成功才算成功。（一半以上投票的机制）

在nacos中，ephemeral默认为true，表示临时实例，使用的是AP架构，只将实例写到内存；非临时实例除了写到内存，还会写到磁盘文件。

Raft中有三种节点状态：

1）follower

2）candidate

3）leader

follower -> candidate：follower向其他节点发送向自己投票的请求，如果超过半数，则认为自己被选举成功；

zookeeper与nacos的 Leader选举的区别：

nacos：raft中每个节点有休眠时间，谁先苏醒，谁先申请投票自己

zookeeper：ZAB每个节点都会进行向自己的投票的动作，然后会进行比对！

所有的修改操作都得在leader节点进行！

raft是一个两阶段写的过程，ZAB也是

1.先写leader，leader广播收到的新数据

2.半数以上返回确认，leader提交到日志，然后将commit同步给follower（日志提交）

选举的超时时间，随机150到300ms之间

先苏醒的节点开始向自己投票，然后告诉其他节点来向自己投票。其他节点都还没开始投票，那就同意这次投票（其他节点重置休眠时间）

新选举的leader向其他节点发送心跳机制（心跳包包含数据）进行同步！（每次心跳都会重置休眠时间）

注意：如果两个节点同时苏醒，那就重新开始，因为票数相同。重新休眠！

节点的数据同步是通过下一次心跳将同步信息同步给follower节点

Raft奇数个节点可以解决集群脑裂的问题。

2.向Leader节点写数据，并同步

RaftConsistencyServiceImpl#put

java">@Overridepublic 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);}}

写新实例，并进行同步

写新实例到leader，然后leader先将该实例写到日志，然后写到leader的内存；

然后leader将数据同步到其他节点，半数以上同步成功，则成功，否则抛出异常。

（小Bug，虽然从节点抛出了异常，但是主节点仍然写成功了，后面的nacos版本通过jraft协议进行了完善）

涉及到的代码如下：

java">public void signalPublish(String key, Record value) throws Exception {if (stopWork) {throw new IllegalStateException("old raft protocol already stop work");}if (!isLeader()) {// 判断当前节点是不是leaderObjectNode 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);// 如果不是leader，则转发到Leader节点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());// leader写完磁盘文件，同步内存final String content = json.toString();final CountDownLatch latch = new CountDownLatch(peers.majorityCount());// leader同步数据for (final String server : peers.allServersIncludeMyself()) {if (isLeader(server)) {latch.countDown();continue;}final String url = buildUrl(server, API_ON_PUB);// nacos原生 只有一步提交// 向其他节点同步数据HttpClient.asyncHttpPostLarge(url, Arrays.asList("key", key), content, new Callback<String>() {@Overridepublic 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();}@Overridepublic void onError(Throwable throwable) {Loggers.RAFT.error("[RAFT] failed to publish data to peer", throwable);}@Overridepublic void onCancel() {}});}// 虽然抛出了异常，但是主节点写成功了// 半数以上写成功了if (!latch.await(UtilsAndCommons.RAFT_PUBLISH_TIMEOUT, TimeUnit.MILLISECONDS)) {// only majority servers return success can we consider this update successLoggers.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();}}

3.集群数据一致性的源码

RaftCore#init()：核心的代码入口

加载数据，写入缓存

包含两个定时任务：选举任务；心跳任务。核心代码如下：

java">@PostConstruct
public void init() throws Exception {Loggers.RAFT.info("initializing Raft sub-system");final long start = System.currentTimeMillis();raftStore.loadDatums(notifier, datums);// 加载数据，写到缓存setTerm(NumberUtils.toLong(raftStore.loadMeta().getProperty("term"), 0L));Loggers.RAFT.info("cache loaded, datum count: {}, current term: {}", datums.size(), peers.getTerm());initialized = true;Loggers.RAFT.info("finish to load data from disk, cost: {} ms.", (System.currentTimeMillis() - start));masterTask = GlobalExecutor.registerMasterElection(new MasterElection());// leader选举任务 500msheartbeatTask = GlobalExecutor.registerHeartbeat(new HeartBeat());// 心跳任务 500msversionJudgement.registerObserver(isAllNewVersion -> {stopWork = isAllNewVersion;if (stopWork) {try {shutdown();raftListener.removeOldRaftMetadata();} catch (NacosException e) {throw new NacosRuntimeException(NacosException.SERVER_ERROR, e);}}}, 100);NotifyCenter.registerSubscriber(notifier);Loggers.RAFT.info("timer started: leader timeout ms: {}, heart-beat timeout ms: {}",GlobalExecutor.LEADER_TIMEOUT_MS, GlobalExecutor.HEARTBEAT_INTERVAL_MS);
}

两个定时任务：

masterTask = GlobalExecutor.registerMasterElection(new MasterElection());// leader选举任务 500ms

heartbeatTask = GlobalExecutor.registerHeartbeat(new HeartBeat());// 心跳任务 500ms

选举的核心代码：

苏醒的节点向其他节点发起投票请求，然后判断是否到达半数。在这个过程中会重置心跳时间，选举时间

用回调函数来判断是否多于半数(用到countdownlatch来控制)，MasterElection#run的核心代码：

java">    @Overridepublic void run() {try {if (stopWork) {return;}if (!peers.isReady()) {return;}RaftPeer local = peers.local();local.leaderDueMs -= GlobalExecutor.TICK_PERIOD_MS;// 随机休眠if (local.leaderDueMs > 0) {return;}// reset timeoutlocal.resetLeaderDue();local.resetHeartbeatDue();sendVote();// 发起投票} catch (Exception e) {Loggers.RAFT.warn("[RAFT] error while master election {}", e);}}

发起选举的核心代码：

java">    private void sendVote() {RaftPeer local = peers.get(NetUtils.localServer());Loggers.RAFT.info("leader timeout, start voting,leader: {}, term: {}", JacksonUtils.toJson(getLeader()),local.term);peers.reset();local.term.incrementAndGet();local.voteFor = local.ip;local.state = RaftPeer.State.CANDIDATE;// 组装投票的信息Map<String, String> params = new HashMap<>(1);params.put("vote", JacksonUtils.toJson(local));for (final String server : peers.allServersWithoutMySelf()) {// 发送投票final String url = buildUrl(server, API_VOTE);try {HttpClient.asyncHttpPost(url, null, params, new Callback<String>() {// 调用http请求@Overridepublic void onReceive(RestResult<String> result) {if (!result.ok()) {Loggers.RAFT.error("NACOS-RAFT vote failed: {}, url: {}", result.getCode(), url);return;}RaftPeer peer = JacksonUtils.toObj(result.getData(), RaftPeer.class);// 投票信息Loggers.RAFT.info("received approve from peer: {}", JacksonUtils.toJson(peer));peers.decideLeader(peer);// 判断其他节点发回来的消息，判断是否多于半数}@Overridepublic void onError(Throwable throwable) {Loggers.RAFT.error("error while sending vote to server: {}", server, throwable);}@Overridepublic void onCancel() {}});} catch (Exception e) {Loggers.RAFT.warn("error while sending vote to server: {}", server);}}}

心跳的核心代码:

只有主节点可以发送心跳，入口方法run->sendBeat

java">@Override
public void run() {try {if (stopWork) {return;}if (!peers.isReady()) {return;}RaftPeer local = peers.local();local.heartbeatDueMs -= GlobalExecutor.TICK_PERIOD_MS;if (local.heartbeatDueMs > 0) {return;}local.resetHeartbeatDue();sendBeat();// 发起心跳} catch (Exception e) {Loggers.RAFT.warn("[RAFT] error while sending beat {}", e);}}

onBeat：将所有注册信息的key和时间戳放到element中，然后进行压缩，然后发给其他节点

RaftController#beat：接收心跳信息

java">@PostMapping("/beat")
public JsonNode beat(HttpServletRequest request, HttpServletResponse response) throws Exception {if (versionJudgement.allMemberIsNewVersion()) {throw new IllegalStateException("old raft protocol already stop");}String entity = new String(IoUtils.tryDecompress(request.getInputStream()), StandardCharsets.UTF_8);String value = URLDecoder.decode(entity, "UTF-8");value = URLDecoder.decode(value, "UTF-8");JsonNode json = JacksonUtils.toObj(value);RaftPeer peer = raftCore.receivedBeat(JacksonUtils.toObj(json.get("beat").asText()));return JacksonUtils.transferToJsonNode(peer);
}

批量处理同步的数据，一批一批的回调leader进行同步；本地删除状态为0的数据（主节点的该数据修改了）。

java">if (batch.size() < 50 && processedCount < beatDatums.size()) {// 批量处理同步数据continue;
}

RaftCore#receivedBeat

java">List<String> deadKeys = new ArrayList<>();
for (Map.Entry<String, Integer> entry : receivedKeysMap.entrySet()) {if (entry.getValue() == 0) {deadKeys.add(entry.getKey());}
}for (String deadKey : deadKeys) {try {deleteDatum(deadKey);} catch (Exception e) {Loggers.RAFT.error("[NACOS-RAFT] failed to remove entry, key={} {}", deadKey, e);}
}