nacos源码6－服务管理－服务端

作者: modou1618 | 来源:发表于2019-01-16 12:23 被阅读0次

一 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

nacos源码6－服务管理－服务端

一 raft

1.1 初始化

1.2 Notifier

1.3 AddressServerUpdater

1.4 HeartBeat

1.4.1 发送心跳包

1.4.2 接收处理心跳包

1.5 MasterElection

1.5.1 选举发送

1.5.2 选举接收处理

二服务管理

2.1 服务实例注册

2.1.1 regDom()

2.1.1 doAddCluster4Dom()

2.1.1 addIp4Dom()

2.2 服务管理DomainsManager

2.2.1 服务变更处理流程

2.2.2 初始化

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

nacos源码6－服务管理－服务端

一 raft

1.1 初始化

1.2 Notifier

1.3 AddressServerUpdater

1.4 HeartBeat

1.4.1 发送心跳包

1.4.2 接收处理心跳包

1.5 MasterElection

1.5.1 选举发送

1.5.2 选举接收处理

二 服务管理

2.1 服务实例注册

2.1.1 regDom()

2.1.1 doAddCluster4Dom()

2.1.1 addIp4Dom()

2.2 服务管理DomainsManager

2.2.1 服务变更处理流程

2.2.2 初始化

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

二服务管理