nacos服务注册之服务器端Raft

Posted 2020-12-21 wenlongliu

Raft是持久化，数据存储在 acosdata amingdata目录
nacos启动后首先从数据存储目录加载数据
Raft协议中节点只有一个LEADER，只有LEADER节点负责数据写入，FOLLOWER节点接受到写入请求后转发给LEADER节点处理
Raft协议中LEADER节点接受写入请求后首先写入本机，然后同步到集群中其他节点，许超过半数节点返回成功，才认为写入成功。
Raft协议中LEADER定时发送心跳数据（包含全量数据）同步FOLLOWER。

Raft存储代码分析； RaftStore类负责数据的存储，数据存储在 acosdata amingdatapublic(namespaceId)目录，
com.alibaba.nacos.naming.domains.meta.public##@@nacos.test.5
com.alibaba.nacos.naming.iplist.public##@@nacos.test.5

@Component
public class RaftStore {

    /**
     * 数据持久化到文件，文件内容就json字符串
     * @param datum
     * @throws Exception
     */
    public synchronized void write(final Datum datum) throws Exception {
        String namespaceId = KeyBuilder.getNamespace(datum.key);
        File cacheFile = new File(cacheDir + File.separator + namespaceId + File.separator + encodeFileName(datum.key));
        FileChannel fc = null;
        ByteBuffer data = ByteBuffer.wrap(JSON.toJSONString(datum).getBytes(StandardCharsets.UTF_8));

        try {
            fc = new FileOutputStream(cacheFile, false).getChannel();
            fc.write(data, data.position());
            fc.force(true);
        } catch (Exception e) {
            MetricsMonitor.getDiskException().increment();
            throw e;
        } finally {
            if (fc != null) {
                fc.close();
            }
        }
    }
}

Raft服务注册源码分析：
只有LEADER节点负责数据写入，FOLLOWER节点接受到写入请求后转发给LEADER节点处理
LEADER节点接受写入请求后首先写入本机，然后同步到集群中其他节点，许超过半数节点返回成功，才认为写入成功。

@Component
public class RaftCore {

    /**
     * 服务注册
     * @param key
     * @param value
     * @throws Exception
     */
    public void signalPublish(String key, Record value) throws Exception {
        //如果是FOLLOWER节点则转发到LEADER节点处理
        if (!isLeader()) {
            JSONObject params = new JSONObject();
            params.put("key", key);
            params.put("value", value);
            Map<String, String> parameters = new HashMap<>(1);
            parameters.put("key", key);
            raftProxy.proxyPostLarge(getLeader().ip, API_PUB, params.toJSONString(), parameters);
            return;
        }
        // LEADER节点处理
        try {
            OPERATE_LOCK.lock();
            final Datum datum = new Datum();
            datum.key = key;
            datum.value = value;
            datum.timestamp.set(getDatum(key).timestamp.incrementAndGet());

            JSONObject json = new JSONObject();
            json.put("datum", datum);
            json.put("source", peers.local());
            //数据注册到本地节点
            onPublish(datum, peers.local());

            final String content = JSON.toJSONString(json);
            //只有大多数服务器（majorityCount=（peers.size() / 2 + 1)）返回成功，我们才能认为这次更新成功
            final CountDownLatch latch = new CountDownLatch(peers.majorityCount());
            //数据同步到集群中的所有节点
            for (final String server : peers.allServersIncludeMyself()) {
                if (isLeader(server)) {
                    latch.countDown();
                    continue;
                }
                // 数据同步地址：/nacos/v1/ns/raft/datum/commit"
                final String url = buildURL(server, API_ON_PUB);
                HttpClient.asyncHttpPostLarge(url, Arrays.asList("key=" + key), content, new AsyncCompletionHandler<Integer>() {
                    @Override
                    public Integer onCompleted(Response response) throws Exception {
                        if (response.getStatusCode() != HttpURLConnection.HTTP_OK) {
                            return 1;
                        }
                        latch.countDown();
                        return 0;
                    }

                    @Override
                    public STATE onContentWriteCompleted() {
                        return STATE.CONTINUE;
                    }
                });

            }
            //等待大多数服务器成功或超时（RAFT_PUBLISH_TIMEOUT=5000）
            if (!latch.await(UtilsAndCommons.RAFT_PUBLISH_TIMEOUT, TimeUnit.MILLISECONDS)) {
                // only majority servers return success can we consider this update success
                throw new IllegalStateException("data publish failed, caused failed to notify majority, key=" + key);
            }
        } finally {
            OPERATE_LOCK.unlock();
        }
    }

    /**
     * 数据注册到本地节点
     * @param datum 
     * @param source    
     * @throws Exception
     */
    public void onPublish(Datum datum, RaftPeer source) throws Exception {
        //验证数据
        .....................
        RaftPeer local = peers.local();
        local.resetLeaderDue();

        // if data should be persisted, usually this is true:
        if (KeyBuilder.matchPersistentKey(datum.key)) {
            raftStore.write(datum);
        }
        // 存入内存（ConcurrentHashMap）
        datums.put(datum.key, datum);

        if (isLeader()) {
            local.term.addAndGet(PUBLISH_TERM_INCREASE_COUNT);
        } else {
            if (local.term.get() + PUBLISH_TERM_INCREASE_COUNT > source.term.get()) {
                //set leader term:
                getLeader().term.set(source.term.get());
                local.term.set(getLeader().term.get());
            } else {
                local.term.addAndGet(PUBLISH_TERM_INCREASE_COUNT);
            }
        }
        //更新任期
        raftStore.updateTerm(local.term.get());
        //通知其他类
        notifier.addTask(datum.key, ApplyAction.CHANGE);
    }
}

Raft协议中LEADER定时(TICK_PERIOD_MS=500毫秒）发送心跳数据（包含全量数据）同步FOLLOWER。

public class HeartBeat implements Runnable {
       /**
         * 发送心跳
         * @throws IOException
         * @throws InterruptedException
         */
        public void sendBeat() throws IOException, InterruptedException {
            RaftPeer local = peers.local();
            if (local.state != RaftPeer.State.LEADER && !STANDALONE_MODE) {
                return;
            }
            local.resetLeaderDue();

            // 构造报文
            JSONObject packet = new JSONObject();
            packet.put("peer", local);
            JSONArray array = new JSONArray();
            if (!switchDomain.isSendBeatOnly()) {
                //遍历所有服务
                for (Datum datum : datums.values()) {
                    JSONObject element = new JSONObject();
                    if (KeyBuilder.matchServiceMetaKey(datum.key)) {
                        element.put("key", KeyBuilder.briefServiceMetaKey(datum.key));
                    } else if (KeyBuilder.matchInstanceListKey(datum.key)) {
                        element.put("key", KeyBuilder.briefInstanceListkey(datum.key));
                    }
                    element.put("timestamp", datum.timestamp);
                    array.add(element);
                }
            }

            packet.put("datums", array);
            // broadcast
            Map<String, String> params = new HashMap<String, String>(1);
            params.put("beat", JSON.toJSONString(packet));
            // 压缩数据
            String content = JSON.toJSONString(params);
            ByteArrayOutputStream out = new ByteArrayOutputStream();
            GZIPOutputStream gzip = new GZIPOutputStream(out);
            gzip.write(content.getBytes(StandardCharsets.UTF_8));
            gzip.close();

            byte[] compressedBytes = out.toByteArray();
            String compressedContent = new String(compressedBytes, StandardCharsets.UTF_8);
            //发送集群中所有节点
            for (final String server : peers.allServersWithoutMySelf()) {
                try {
                    final String url = buildURL(server, API_BEAT);
                    HttpClient.asyncHttpPostLarge(url, null, compressedBytes, new AsyncCompletionHandler<Integer>() {
                        @Override
                        public Integer onCompleted(Response response) throws Exception {
                            if (response.getStatusCode() != HttpURLConnection.HTTP_OK) {
                                MetricsMonitor.getLeaderSendBeatFailedException().increment();
                                return 1;
                            }

                            peers.update(JSON.parseObject(response.getResponseBody(), RaftPeer.class));
                            return 0;
                        }
                    });
                } catch (Exception e) {
                    Loggers.RAFT.error("error while sending heart-beat to peer: {} {}", server, e);
                    MetricsMonitor.getLeaderSendBeatFailedException().increment();
                }
            }

        }
}

Raft协议选举流程：
nacos启动时启动一个选举定时任务：executorService.scheduleAtFixedRate(runnable, 0, TICK_PERIOD_MS=500L, TimeUnit.MILLISECONDS);
nacos节点定时任务检测如果超过15秒没有收到LEADER心跳则发起选举投票（选自己为LEADER），发送到集群其他节点，自己状态为CANDIDATE。
nacos节点收到选举投票如果CANDIDATE节点term大于本地的term则同意发送节点为LEADER，否则投票自己为LEADER。
CANDIDATE节点依次收到其他节点的投票回复，统计投票，只要某个节点超过半数投票则确认为LEADER。
LEADER节点同过心跳通知其他节点，自己为新LEADER。

    public class MasterElection implements Runnable {
        @Override
        public void run() {
            try {
                if (!peers.isReady()) {
                    return;
                }

                RaftPeer local = peers.local();
                local.leaderDueMs -= GlobalExecutor.TICK_PERIOD_MS;
                //是否超过15秒没有收到LEADER心跳
                if (local.leaderDueMs > 0) {
                    return;
                }
                // reset timeout
                local.resetLeaderDue();
                local.resetHeartbeatDue();
                // 发送选举
                sendVote();
            } catch (Exception e) {
                Loggers.RAFT.warn("[RAFT] error while master election {}", e);
            }

        }

        /**
         * 发送选举流程
         */
        public void sendVote() {
            RaftPeer local = peers.get(NetUtils.localServer());
            peers.reset();
            local.term.incrementAndGet();
            local.voteFor = local.ip; //选自己
            local.state = RaftPeer.State.CANDIDATE;

            Map<String, String> params = new HashMap<>(1);
            params.put("vote", JSON.toJSONString(local));
            //发送集群其他节点: /nacos/v1/ns/raft/vote
            for (final String server : peers.allServersWithoutMySelf()) {
                final String url = buildURL(server, API_VOTE);
                try {
                    HttpClient.asyncHttpPost(url, null, params, new AsyncCompletionHandler<Integer>() {
                        @Override
                        public Integer onCompleted(Response response) throws Exception {
                            //收到回复统计选票确定谁是LEADER
                            RaftPeer peer = JSON.parseObject(response.getResponseBody(), RaftPeer.class);
                            peers.decideLeader(peer);
                            return 0;
                        }
                    });
                } catch (Exception e) {
                    Loggers.RAFT.warn("error while sending vote to server: {}", server);
                }
            }
        }
    }

收到选举投票请求的处理

   /**
     * 收到选举投票请求
     * @param remote  CANDIDATE节点
     * @return        自己投票的节点
     */
    public synchronized RaftPeer receivedVote(RaftPeer remote) {
        RaftPeer local = peers.get(NetUtils.localServer());
        // 本地节点term大于等于CANDIDATE节点term则投票自己为LEADER
        if (remote.term.get() <= local.term.get()) {
            if (StringUtils.isEmpty(local.voteFor)) {
                local.voteFor = local.ip;
            }
            return local;
        }
        // CANDIDATE节点term大于本地的term则同意CANDIDATE节点为LEADER
        local.resetLeaderDue();
        local.state = RaftPeer.State.FOLLOWER;
        local.voteFor = remote.ip;
        local.term.set(remote.term.get());
        return local;
    }

收到投票回复统计选票确定谁是LEADER

    /**
     * 统计选票确定谁是LEADER
     * @param candidate  一次计票
     * @return
     */
    public RaftPeer decideLeader(RaftPeer candidate) {
        //放到投票箱
        peers.put(candidate.ip, candidate);
        SortedBag ips = new TreeBag();
        int maxApproveCount = 0;
        String maxApprovePeer = null;
        //统计投票找出最大的投票节点
        for (RaftPeer peer : peers.values()) {
            if (StringUtils.isEmpty(peer.voteFor)) {
                continue;
            }
            ips.add(peer.voteFor);
            if (ips.getCount(peer.voteFor) > maxApproveCount) {
                maxApproveCount = ips.getCount(peer.voteFor);
                maxApprovePeer = peer.voteFor;
            }
        }
        // 只要超过半数投票则确认为LEADER。
        if (maxApproveCount >= majorityCount()) {
            RaftPeer peer = peers.get(maxApprovePeer);
            peer.state = RaftPeer.State.LEADER;
            if (!Objects.equals(leader, peer)) {
                leader = peer;
                applicationContext.publishEvent(new LeaderElectFinishedEvent(this, leader));
            }
        }
        return leader;
    }