一 raft

1.1 初始化

• 监听ApplicationEvent事件，调用初始化函数

public void onApplicationEvent(WebServerInitializedEvent event) {
    RaftCore.init();
}

• 线程池执行服务变更通知任务executor.submit(notifier);
• PeerSet peers存储raft节点信息，添加所有服务管理服务端为raft节点peers.add(NamingProxy.getServers());
• RaftStore.load();加载文件缓存数据
• 注册周期调用函数，leader选举，leader心跳发送，raft节点列表更新

GlobalExecutor.register(new MasterElection());
GlobalExecutor.register1(new HeartBeat());
GlobalExecutor.register(new AddressServerUpdater(), GlobalExecutor.ADDRESS_SERVER_UPDATE_INTERVAL_MS);

• 加锁设置初始化完成标记，加锁原因暂时未看明白

if (lock.tryLock(INIT_LOCK_TIME_SECONDS, TimeUnit.SECONDS)) {
    initialized = true;
    lock.unlock();
}

1.2 Notifier

• BlockingQueue<Pair> tasks 支持异步处理服务变更消息
• ConcurrentHashMap<String, String> services 编码重复的服务change消息处理
• 遍历List<RaftListener> listeners监听列表，选择listener.interests(datum.key)监听当前变更的服务的listener，触发对应的onChange服务变更处理或onDelete服务删除处理。
• 添加服务变更监听函数，ConcurrentMap<String, Datum> datums中若有当前监听interests的服务，则立即调用onChange回调通知

public static void listen(RaftListener listener) {
    if (listeners.contains(listener)) {
        return;
    }

    listeners.add(listener);

    for (RaftListener listener1 : listeners) {
        if (listener1 instanceof VirtualClusterDomain) {
            Loggers.RAFT.debug("listener in listeners: {}", ((VirtualClusterDomain) listener1).getName());
        }
    }

    // if data present, notify immediately
    for (Datum datum : datums.values()) {
        if (!listener.interests(datum.key)) {
            continue;
        }

        try {
            listener.onChange(datum.key, datum.value);
        } catch (Exception e) {
            Loggers.RAFT.error("NACOS-RAFT failed to notify listener", e);
        }
    }
}

1.3 AddressServerUpdater

• 实时获取服务管理server地址列表，和当前raft节点对比，删除下线节点，添加新增节点。

List<String> servers = NamingProxy.getServers();
List<RaftPeer> peerList = new ArrayList<RaftPeer>(peers.allPeers());

1.4 HeartBeat

1.4.1 发送心跳包

• local.resetHeartbeatDue();设置心跳周期5s
• sendBeat(); 发送心跳报文

public void run() {
    try {
        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);
    }

}

• 仅集群leader发送心跳

RaftPeer local = peers.local();
if (local.state != RaftPeer.State.LEADER && !STANDALONE_MODE) {
    return;
}

• local.resetLeaderDue();重置leader周期15s－20s
• 组装心跳报文，本节点信息，服务信息，
• Switch.isSendBeatOnly()表示心跳报文是否包含服务信息，服务信息包含
  RAFT_DOM_PRE = "meta";
  RAFT_IPLIST_PRE = "iplist.";
  RAFT_TAG_DOM_PRE = "tag.meta";
  RAFT_TAG_IPLIST_PRE = "tag.iplist.";

// build data
JSONObject packet = new JSONObject();
packet.put("peer", local);

JSONArray array = new JSONArray();

if (Switch.isSendBeatOnly()) {
    Loggers.RAFT.info("[SEND-BEAT-ONLY] {}", String.valueOf(Switch.isSendBeatOnly()));
}

if (!Switch.isSendBeatOnly()) {
    for (Datum datum : datums.values()) {

        JSONObject element = new JSONObject();
        String key;

        if (datum.key.startsWith(UtilsAndCommons.DOMAINS_DATA_ID)) {
            key = (datum.key).split(UtilsAndCommons.DOMAINS_DATA_ID)[1];
            element.put("key", UtilsAndCommons.RAFT_DOM_PRE + key);
        } else if (datum.key.startsWith(UtilsAndCommons.IPADDRESS_DATA_ID_PRE)) {
            key = (datum.key).split(UtilsAndCommons.IPADDRESS_DATA_ID_PRE)[1];
            element.put("key", UtilsAndCommons.RAFT_IPLIST_PRE + key);
        } else if (datum.key.startsWith(UtilsAndCommons.TAG_DOMAINS_DATA_ID)) {
            key = (datum.key).split(UtilsAndCommons.TAG_DOMAINS_DATA_ID)[1];
            element.put("key", UtilsAndCommons.RAFT_TAG_DOM_PRE + key);
        } else if (datum.key.startsWith(UtilsAndCommons.NODE_TAG_IP_PRE)) {
            key = (datum.key).split(UtilsAndCommons.NODE_TAG_IP_PRE)[1];
            element.put("key", UtilsAndCommons.RAFT_TAG_IPLIST_PRE + key);
        }
        element.put("timestamp", datum.timestamp);

        array.add(element);
    }
} else {
    Loggers.RAFT.info("[RAFT] send beat only.");
}

packet.put("datums", array);
// broadcast
Map<String, String> params = new HashMap<String, String>(1);
params.put("beat", JSON.toJSONString(packet));

• GZIPOutputStream gzip = new GZIPOutputStream(out);使用gzip压缩方式发送
• 遍历其他节点peers.allServersWithoutMySelf()，使用HttpClient.asyncHttpPostLarge异步发送心跳报文
• 异步报文完成回调onCompleted，更新本地缓存的对端raft节点信息

public Integer onCompleted(Response response) throws Exception {
    if (response.getStatusCode() != HttpURLConnection.HTTP_OK) {
        Loggers.RAFT.error("NACOS-RAFT beat failed: {}, peer: {}",
            response.getResponseBody(), server);
        MetricsMonitor.getLeaderSendBeatFailedException().increment();
        return 1;
    }

    peers.update(JSON.parseObject(response.getResponseBody(), RaftPeer.class));
    Loggers.RAFT.info("receive beat response from: {}", url);
    return 0;
}

1.4.2 接收处理心跳包

• 心跳http请求处理

@RequestMapping("/beat")
public JSONObject beat(HttpServletRequest request, HttpServletResponse response) throws Exception {

    String entity = new String(IoUtils.tryDecompress(request.getInputStream()), "UTF-8");

    String value = Arrays.asList(entity).toArray(new String[1])[0];

    JSONObject json = JSON.parseObject(value);
    JSONObject beat = JSON.parseObject(json.getString("beat"));

    RaftPeer peer = RaftCore.HeartBeat.receivedBeat(beat);

    return JSON.parseObject(JSON.toJSONString(peer));
}

• raft节点状态为RaftPeer.State.LEADER才可发送心跳报文
• 节点term最大的才可为leader，若leader节点的term小于本节点term，则表示leader选举错误
• 更新本节点状态为RaftPeer.State.FOLLOWER，设置本节点voteFor投票leader为当前leader
• local.resetLeaderDue();重置leader周期15s-20s
• local.resetHeartbeatDue() 重置心跳周期5s
• makeLeader()更新PeerSet.leader为当前leader，获取其他状态为leader的节点信息更新本地缓存。
• Switch.isSendBeatOnly()心跳报文协议服务信息，则进行更新判断
• 获取服务信息及对应时间戳

JSONObject entry = (JSONObject) object;
String key = entry.getString("key");
final String datumKey;

if (key.startsWith(UtilsAndCommons.RAFT_DOM_PRE)) {
    int index = key.indexOf(UtilsAndCommons.RAFT_DOM_PRE);
    datumKey = UtilsAndCommons.DOMAINS_DATA_ID + key.substring(index + UtilsAndCommons.RAFT_DOM_PRE.length());
} else if (key.startsWith(UtilsAndCommons.RAFT_IPLIST_PRE)) {
    int index = key.indexOf(UtilsAndCommons.RAFT_IPLIST_PRE);
    datumKey = UtilsAndCommons.IPADDRESS_DATA_ID_PRE + key.substring(index + UtilsAndCommons.RAFT_IPLIST_PRE.length());
} else if (key.startsWith(UtilsAndCommons.RAFT_TAG_DOM_PRE)) {
    int index = key.indexOf(UtilsAndCommons.RAFT_TAG_DOM_PRE);
    datumKey = UtilsAndCommons.TAG_DOMAINS_DATA_ID + key.substring(index + UtilsAndCommons.RAFT_TAG_DOM_PRE.length());
} else {
    int index = key.indexOf(UtilsAndCommons.RAFT_TAG_IPLIST_PRE);
    datumKey = UtilsAndCommons.NODE_TAG_IP_PRE + key.substring(index + UtilsAndCommons.RAFT_TAG_IPLIST_PRE.length());
}

long timestamp = entry.getLong("timestamp");

receivedKeysMap.put(datumKey, 1);

• 新增服务或服务时间戳小于leader上的服务时间戳，则做更新，一次最多50条纪录进行更新。

if (RaftCore.datums.containsKey(datumKey) && RaftCore.datums.get(datumKey).timestamp.get() >= timestamp && processedCount < beatDatums.size()) {
    continue;
}

if (!(RaftCore.datums.containsKey(datumKey) && RaftCore.datums.get(datumKey).timestamp.get() >= timestamp)) {
    batch.add(datumKey);
}

if (batch.size() < 50 && processedCount < beatDatums.size()) {
    continue;
}

• HttpClient.asyncHttpGet异步http请求获取leader上这批服务的详细信息，最多50个。
• OPERATE_LOCK.lock(); 加锁控制并发更新服务信息
• 元数据信息持久化到文件，或配置了INSTANCE_LIST_PERSISTED就持久化实例信息到文件中

if (datum.key.startsWith(UtilsAndCommons.DOMAINS_DATA_ID) ||
    UtilsAndCommons.INSTANCE_LIST_PERSISTED) {
    RaftStore.write(datum);
}

• 更新内存服务信息缓存，重置leader周期时间
• 服务元数据信息，则更新local.term＋100，若大于leader.term则更新local.term为lead.term。
• notifier.addTask(datum, Notifier.ApplyAction.CHANGE); 通知服务变更监听函数
• 每批更新休眠200ms。
• 删除leader中不存在的服务信息，包括内存和文件缓存。

1.5 MasterElection

1.5.1 选举发送

• leader周期超时未更新后，开启leader选举
• 一轮选举不成功，则因为更新了leader周期，则会等待15s-20s后重新选举。
• 清空各节点的投票对象voteFor，本节点投票给自己，修改状态为leader候选者，选举时每轮term+1。进行100轮选举后才会比数据变更的term大。

peers.reset();
local.term.incrementAndGet();
local.voteFor = local.ip;
local.state = RaftPeer.State.CANDIDATE;

• 遍历所有raft节点，异步发送选举报文，decideLeader()根据返回结果确认leader
  每个选举报文通信，获取目标节点的投票信息，更新本地缓存
  遍历raft节点，投票数超过majorityCount()的节点即为新的leader节点

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;
        }
    }

    if (maxApproveCount >= majorityCount()) {
        RaftPeer peer = peers.get(maxApprovePeer);
        peer.state = RaftPeer.State.LEADER;

        if (!Objects.equals(leader, peer)) {
            leader = peer;
            Loggers.RAFT.info("{} has become the LEADER", leader.ip);
        }
    }

    return leader;
}

1.5.2 选举接收处理

• 比较term值，本节点term大，投票本节点，返回本节点信息。

local.voteFor = local.ip;
return local;

• 本节点term小，则投票来源节点，更新本节点为follow状态。

local.resetLeaderDue();
local.state = RaftPeer.State.FOLLOWER;
local.voteFor = remote.ip;
local.term.set(remote.term.get());
return local;

二 服务管理

2.1 服务实例注册

服务实例注册.png

2.1.1 regDom()

• 新增服务信息

  
  
  regDom.png

2.1.1 doAddCluster4Dom()

• 变更服务集群

  
  
  doAddCluster4Dom.png

2.1.1 addIp4Dom()

• 变更服务实例

  
  
  变更服务实例.png

2.2 服务管理DomainsManager

2.2.1 服务变更处理流程

变更.png

2.2.2 初始化

初始化.png