MetaServer 的SOFARegistry是什么

本篇文章為大家展示了MetaServer 之什么是SOFARegistry，內(nèi)容簡(jiǎn)明扼要并且容易理解，絕對(duì)能使你眼前一亮，通過這篇文章的詳細(xì)介紹希望你能有所收獲。

功能介紹

MetaServer 作為 SOFARegistry 的元數(shù)據(jù)中心，其核心功能可以概括為集群成員管理。分布式系統(tǒng)中，如何知道集群中有哪些節(jié)點(diǎn)列表，如何處理集群擴(kuò)所容，如何處理集群節(jié)點(diǎn)異常，都是不得不考慮的問題。MetaServer 的存在就是解決這些問題，其在 SOFARegistry 中位置如圖所示： cdn.nlark.com/yuque/0/2019/png/338467/1568254454389-0cefa85d-131a-4c2d-a844-f66e2c9807b4.png">

MetaServer 通過 SOFAJRaft 保證高可用和一致性，類似于注冊(cè)中心，管理著集群內(nèi)部的成員列表：

• 節(jié)點(diǎn)列表的注冊(cè)與存儲(chǔ)

• 節(jié)點(diǎn)列表的變更通知

• 節(jié)點(diǎn)健康監(jiān)測(cè)

內(nèi)部架構(gòu)

內(nèi)部架構(gòu)如下圖所示：

MetaServer 基于 Bolt, 通過 TCP 私有協(xié)議的形式對(duì)外提供服務(wù)，包括 DataServer, SessionServer 等，處理節(jié)點(diǎn)的注冊(cè)，續(xù)約和列表查詢等請(qǐng)求。

同時(shí)也基于 Http 協(xié)議提供控制接口，比如可以控制 session 節(jié)點(diǎn)是否開啟變更通知, 健康檢查接口等。

成員列表數(shù)據(jù)存儲(chǔ)在 Repository 中，Repository 被一致性協(xié)議層進(jìn)行包裝，作為 SOFAJRaft 的狀態(tài)機(jī)實(shí)現(xiàn)，所有對(duì) Repository 的操作都會(huì)同步到其他節(jié)點(diǎn), 通過Rgistry來操作存儲(chǔ)層。

MetaServer 使用 Raft 協(xié)議保證數(shù)據(jù)一致性， 同時(shí)也會(huì)保持與注冊(cè)的節(jié)點(diǎn)的心跳，對(duì)于心跳超時(shí)沒有續(xù)約的節(jié)點(diǎn)進(jìn)行驅(qū)逐，來保證數(shù)據(jù)的有效性。

在可用性方面，只要未超過半數(shù)節(jié)點(diǎn)掛掉，集群都可以正常對(duì)外提供服務(wù)， 半數(shù)以上掛掉，Raft 協(xié)議無法選主和日志復(fù)制，因此無法保證注冊(cè)的成員數(shù)據(jù)的一致性和有效性。整個(gè)集群不可用 不會(huì)影響 Data 和 Session 節(jié)點(diǎn)的正常功能，只是無法感知節(jié)點(diǎn)列表變化。

源碼分析

服務(wù)啟動(dòng)

MetaServer 在啟動(dòng)時(shí)，會(huì)啟動(dòng)三個(gè) Bolt Server，并且注冊(cè) Processor Handler，處理對(duì)應(yīng)的請(qǐng)求, 如下圖所示：

• DataServer：處理 DataNode 相關(guān)的請(qǐng)求；

• SessionServer：處理 SessionNode 相關(guān)的請(qǐng)求；

• MetaServer：處理MetaNode相關(guān)的請(qǐng)求；

然后啟動(dòng) HttpServer, 用于處理 Admin 請(qǐng)求，提供推送開關(guān)，集群數(shù)據(jù)查詢等 Http 接口。

最后啟動(dòng) Raft 服務(wù)， 每個(gè)節(jié)點(diǎn)同時(shí)作為 RaftClient 和 RaftServer, 用于集群間的變更和數(shù)據(jù)同步。

各個(gè) Server 的默認(rèn)端口分別為：

meta.server.sessionServerPort=9610
meta.server.dataServerPort=9611
meta.server.metaServerPort=9612
meta.server.raftServerPort=9614
meta.server.httpServerPort=9615

節(jié)點(diǎn)注冊(cè)

由上節(jié)可知，DataServer 和 SessionServer 都有處理節(jié)點(diǎn)注冊(cè)請(qǐng)求的 Handler。注冊(cè)行為由 Registry 完成。注冊(cè)接口實(shí)現(xiàn)為：

@Override
    public NodeChangeResult register(Node node) {
        StoreService storeService =          ServiceFactory.getStoreService(node.getNodeType());
        return storeService.addNode(node);
    }

Regitsry 根據(jù)不同的節(jié)點(diǎn)類型，獲取對(duì)應(yīng)的StoreService，比如DataNode，其實(shí)現(xiàn)為 DataStoreService 然后由 StoreService  存儲(chǔ)到 Repository  中，具體實(shí)現(xiàn)為：

// 存儲(chǔ)節(jié)點(diǎn)信息
dataRepositoryService.put(ipAddress, new RenewDecorate(dataNode, RenewDecorate.DEFAULT_DURATION_SECS));
//...
// 存儲(chǔ)變更事件
dataConfirmStatusService.putConfirmNode(dataNode, DataOperator.ADD);

調(diào)用 RepositoryService#put  接口存儲(chǔ)后，同時(shí)會(huì)存儲(chǔ)一個(gè)變更事件到隊(duì)列中，主要用于數(shù)據(jù)推送，消費(fèi)處理。

節(jié)點(diǎn)數(shù)據(jù)的存儲(chǔ)，其本質(zhì)上是存儲(chǔ)在內(nèi)存的哈希表中，其存儲(chǔ)結(jié)構(gòu)為：

// RepositoryService 底層存儲(chǔ)
Map<String/*dataCenter*/, NodeRepository> registry;

// NodeRepository 底層存儲(chǔ)
Map<String/*ipAddress*/, RenewDecorate<T>> nodeMap;

將RenewDecorate存儲(chǔ)到該 Map 中，整個(gè)節(jié)點(diǎn)注冊(cè)的流程就完成了，至于如何和 Raft 協(xié)議進(jìn)行結(jié)合和數(shù)據(jù)同步，下文介紹。

節(jié)點(diǎn)移除的邏輯類似，將節(jié)點(diǎn)信息從該 Map 中刪除，也會(huì)存儲(chǔ)一個(gè)變更事件到隊(duì)列。

注冊(cè)信息續(xù)約和驅(qū)逐

不知道有沒有注意到，節(jié)點(diǎn)注冊(cè)的時(shí)候，節(jié)點(diǎn)信息被 RenewDecorate  包裝起來了，這個(gè)就是實(shí)現(xiàn)注冊(cè)信息續(xù)約和驅(qū)逐的關(guān)鍵：

    private T               renewal;  // 節(jié)點(diǎn)對(duì)象封裝
    private long            beginTimestamp; // 注冊(cè)事件
    private volatile long   lastUpdateTimestamp; // 續(xù)約時(shí)間
    private long            duration; // 超時(shí)時(shí)間

該對(duì)象為注冊(cè)節(jié)點(diǎn)信息，附加了注冊(cè)時(shí)間、上次續(xù)約時(shí)間、過期時(shí)間。那么續(xù)約操作就是修改lastUpdateTimestamp，是否過期就是判斷System.currentTimeMillis() - lastUpdateTimestamp > duration 是否成立，成立則認(rèn)為節(jié)點(diǎn)超時(shí)進(jìn)行驅(qū)逐。

和注冊(cè)一樣，續(xù)約請(qǐng)求的處理 Handler 為ReNewNodesRequestHandler，最終交由 StoreService 進(jìn)行續(xù)約操作。另外一點(diǎn)，續(xù)約的時(shí)候如果沒有查詢到注冊(cè)節(jié)點(diǎn)，會(huì)觸發(fā)節(jié)點(diǎn)注冊(cè)的操作。

驅(qū)出的操作是由定時(shí)任務(wù)完成，MetaServer 在啟動(dòng)時(shí)會(huì)啟動(dòng)多個(gè)定時(shí)任務(wù)，詳見ExecutorManager#startScheduler,，其中一個(gè)任務(wù)會(huì)調(diào)用Registry#evict，其實(shí)現(xiàn)為遍歷存儲(chǔ)的 Map, 獲得過期的列表，調(diào)用StoreService#removeNodes方法，將他們從 Repository  中移除，這個(gè)操作也會(huì)觸發(fā)變更通知。該任務(wù)默認(rèn)每3秒執(zhí)行一次。

節(jié)點(diǎn)列表變更推送

上文有介紹到，在處理節(jié)點(diǎn)注冊(cè)請(qǐng)求后，也會(huì)存儲(chǔ)一個(gè)節(jié)點(diǎn)變更事件，即：

dataConfirmStatusService.putConfirmNode(dataNode, DataOperator.ADD);

DataConfirmStatusService  也是一個(gè)由 Raft 協(xié)議進(jìn)行同步的存儲(chǔ)，其存儲(chǔ)結(jié)構(gòu)為：

BlockingQueue<NodeOperator>  expectNodesOrders = new LinkedBlockingQueue();

ConcurrentHashMap<DataNode/*node*/, Map<String/*ipAddress*/, DataNode>> expectNodes = new ConcurrentHashMap<>();

• expectNodesOrders 用來存儲(chǔ)節(jié)點(diǎn)變更事件；

• expectNodes 用來存儲(chǔ)變更事件需要確認(rèn)的節(jié)點(diǎn)，也就是說 NodeOperator  只有得到了其他節(jié)點(diǎn)的確認(rèn)，才會(huì)從 expectNodesOrders 移除；

那么事件存儲(chǔ)到 BlockingQueue 里，哪里去消費(fèi)呢？ 看源碼發(fā)現(xiàn)，并不是想象中的使用一個(gè)線程阻塞的讀。

在ExecutorManager中會(huì)啟動(dòng)一個(gè)定時(shí)任務(wù)，輪詢?cè)撽?duì)列有沒有數(shù)據(jù)。即周期性的調(diào)用Registry#pushNodeListChange方法，獲取隊(duì)列的頭節(jié)點(diǎn)并消費(fèi)。Data 和 Session 各對(duì)應(yīng)一個(gè)任務(wù)。具體流程如下圖所示：

1. 首先獲取隊(duì)列（expectNodesOrders）頭節(jié)點(diǎn)，如果為Null直接返回；

2. 獲取當(dāng)前數(shù)據(jù)中心的節(jié)點(diǎn)列表，并存儲(chǔ)到確認(rèn)表（expectNodes）；

3. 提交節(jié)點(diǎn)變更推送任務(wù)（firePushXxListTask)；

4. 處理任務(wù)，即調(diào)用 XxNodeService 的 pushXxxNode 方法，即通過 ConnectionHandler 獲取所有的節(jié)點(diǎn)連接，發(fā)送節(jié)點(diǎn)列表；

5. 收到回復(fù)后，如果需要確認(rèn)，則會(huì)調(diào)用StroeService#confirmNodeStatus 方法，將該節(jié)點(diǎn)從expectNodes中移除；

6. 待所有的節(jié)點(diǎn)從 expectNodes 中移除，則將此次操作從 expectNodesOrders 移除，處理完畢；

節(jié)點(diǎn)列表查詢

Data，Meta,Session Server 都提供 getNodesRequestHandler ，用于處理查詢當(dāng)前節(jié)點(diǎn)列表的請(qǐng)求，其本質(zhì)上從底層存儲(chǔ) Repository 讀取數(shù)據(jù)返回，這里不在贅述。返回的結(jié)果的具體結(jié)構(gòu)見 NodeChangeResult 類，包含各個(gè)數(shù)據(jù)中心的節(jié)點(diǎn)列表以及版本號(hào)。

基于 Raft 的存儲(chǔ)

后端 Repository 可以看作SOFAJRaft 的狀態(tài)機(jī)，任何對(duì) Map 的操作都會(huì)在集群內(nèi)部，交由 Raft 協(xié)議進(jìn)行同步，從而達(dá)到集群內(nèi)部的一致。從源碼上看，所有的操作都是直接調(diào)用的 RepositoryService 等接口，那么是如何和 Raft 服務(wù)結(jié)合起來的呢？

看源碼會(huì)發(fā)現(xiàn)，凡是引用 RepositoryService 的地方，都加了 @RaftReference， RepositoryService 的具體實(shí)現(xiàn)類都加了 @RaftService 注解。其關(guān)鍵就在這里，其處理類為 RaftAnnotationBeanPostProcessor。具體流程如下：

在 processRaftReference  方法中，凡是加了 @RaftReference 注解的屬性，都會(huì)被動(dòng)態(tài)代理類替換，其代理實(shí)現(xiàn)見 ProxyHandler 類，即將方法調(diào)用，封裝為 ProcessRequest，通過 RaftClient 發(fā)送給 RaftServer。

而被加了 @RaftService 的類會(huì)被添加到 Processor 類 中，通過 serviceId(interfaceName + uniqueId) 進(jìn)行區(qū)分。RaftServer 收到請(qǐng)求后，會(huì)把它生效到 SOFAJRaft 的狀態(tài)機(jī)，具體實(shí)現(xiàn)類為 ServiceStateMachine，即會(huì)調(diào)用 Processor 方法，通過 serviceId 找到這個(gè)實(shí)現(xiàn)類，執(zhí)行對(duì)應(yīng)的方法調(diào)用。

當(dāng)然如果本機(jī)就是主節(jié)點(diǎn), 對(duì)于一些查詢請(qǐng)求不需要走Raft協(xié)議而直接調(diào)用本地實(shí)現(xiàn)方法。 

這個(gè)過程其實(shí)和 RPC 調(diào)用非常類似，在引用方發(fā)起的方法調(diào)用，并不會(huì)真正的執(zhí)行方法，而是封裝成請(qǐng)求發(fā)送到 Raft 服務(wù)，由 Raft 狀態(tài)機(jī)進(jìn)行真正的方法調(diào)用，比如把節(jié)點(diǎn)信息存儲(chǔ)到 Map 中。所有節(jié)點(diǎn)之間的數(shù)據(jù)一致由Raft協(xié)議進(jìn)行保證。

總結(jié)

在分布式系統(tǒng)中，集群成員管理是避不開的問題，有些集群直接把列表信息寫到配置文件或者配置中心，也有的集群選擇使用 zookeeper 或者 etcd 等維護(hù)集群元數(shù)據(jù)，SOFARegistry 選擇基于一致性協(xié)議 Raft，開發(fā)獨(dú)立的MetaServer，來實(shí)現(xiàn)集群列表維護(hù)和變更實(shí)時(shí)推送，以提高集群管理的靈活性和集群的健壯性。

上述內(nèi)容就是MetaServer 之什么是SOFARegistry，你們學(xué)到知識(shí)或技能了嗎？如果還想學(xué)到更多技能或者豐富自己的知識(shí)儲(chǔ)備，歡迎關(guān)注億速云行業(yè)資訊頻道。