螞蟻金服服務註冊中心 MetaServer 功能介紹和實現剖析 | SOFARegistry 解析

SOFAStack （Scalable Open Financial  Architecture Stack） 是螞蟻金服自主研發的金融級分佈式架構，包含了構建金融級雲原生架構所需的各個組件，是在金融場景裏錘鍊出來的最佳實踐。

SOFARegistry 是螞蟻金服開源的具備承載海量服務註冊和訂閱能力的、高可用的服務註冊中心，在支付寶/螞蟻金服的業務發展驅動下，近十年間已經演進至第五代。

本文爲《剖析 | SOFARegistry 框架》第三篇，本篇做者 Yavin，來自考拉海購。《剖析 | SOFARegistry 框架》系列由 SOFA 團隊和源碼愛好者們出品，項目代號：SOFA:RegistryLab/，文末包含往期系列文章。

GitHub 地址：https://github.com/sofastack/sofa-registry

導讀

集羣成員管理是分佈式系統中繞不開的話題。MetaServer 在 SOFARegistry 中，承擔着集羣元數據管理的角色，用來維護集羣成員列表。本文但願從 MetaServer 的功能和部分源碼切入剖析，爲學習研究、或者項目中使用SOFARegistry 的開發者帶來一些啓發，分爲三個部分：

• 功能介紹
• 內部架構
• 源碼分析

功能介紹

MetaServer 做爲 SOFARegistry 的元數據中心，其核心功能能夠歸納爲集羣成員管理。分佈式系統中，如何知道集羣中有哪些節點列表，如何處理集羣擴所容，如何處理集羣節點異常，都是不得不考慮的問題。MetaServer 的存在就是解決這些問題，其在 SOFARegistry 中位置如圖所示：

MetaServer 經過 SOFAJRaft 保證高可用和一致性，相似於註冊中心，管理着集羣內部的成員列表：

• 節點列表的註冊與存儲
• 節點列表的變動通知
• 節點健康監測

內部架構

內部架構以下圖所示：

MetaServer 基於 Bolt, 經過 TCP 私有協議的形式對外提供服務，包括 DataServer, SessionServer 等，處理節點的註冊，續約和列表查詢等請求。

同時也基於 Http 協議提供控制接口，好比能夠控制 session 節點是否開啓變動通知, 健康檢查接口等。

成員列表數據存儲在 Repository 中，Repository 被一致性協議層進行包裝，做爲 SOFAJRaft 的狀態機實現，全部對 Repository 的操做都會同步到其餘節點, 經過Rgistry來操做存儲層。

MetaServer 使用 Raft 協議保證數據一致性， 同時也會保持與註冊的節點的心跳，對於心跳超時沒有續約的節點進行驅逐，來保證數據的有效性。

在可用性方面，只要未超過半數節點掛掉，集羣均可以正常對外提供服務， 半數以上掛掉，Raft 協議沒法選主和日誌複製，所以沒法保證註冊的成員數據的一致性和有效性。整個集羣不可用 不會影響 Data 和 Session 節點的正常功能，只是沒法感知節點列表變化。

源碼分析

服務啓動

MetaServer 在啓動時，會啓動三個 Bolt Server，而且註冊 Processor Handler，處理對應的請求, 以下圖所示：

• DataServer：處理 DataNode 相關的請求；
• SessionServer：處理 SessionNode 相關的請求；
• MetaServer：處理MetaNode相關的請求；

而後啓動 HttpServer, 用於處理 Admin 請求，提供推送開關，集羣數據查詢等 Http 接口。

最後啓動 Raft 服務， 每一個節點同時做爲 RaftClient 和 RaftServer, 用於集羣間的變動和數據同步。

各個 Server 的默認端口分別爲：

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

節點註冊

由上節可知，DataServer 和 SessionServer 都有處理節點註冊請求的 Handler。註冊行爲由 Registry 完成。註冊接口實現爲：

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

Regitsry 根據不一樣的節點類型，獲取對應的StoreService，好比DataNode，其實現爲 DataStoreService 而後由 StoreService  存儲到 Repository  中，具體實現爲：

// 存儲節點信息
dataRepositoryService.put(ipAddress, new RenewDecorate(dataNode, RenewDecorate.DEFAULT_DURATION_SECS));
//...
// 存儲變動事件
dataConfirmStatusService.putConfirmNode(dataNode, DataOperator.ADD);

調用 RepositoryService#put  接口存儲後，同時會存儲一個變動事件到隊列中，主要用於數據推送，消費處理。

節點數據的存儲，其本質上是存儲在內存的哈希表中，其存儲結構爲：

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

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

將RenewDecorate存儲到該 Map 中，整個節點註冊的流程就完成了，至於如何和 Raft 協議進行結合和數據同步，下文介紹。

節點移除的邏輯相似，將節點信息從該 Map 中刪除，也會存儲一個變動事件到隊列。

註冊信息續約和驅逐

不知道有沒有注意到，節點註冊的時候，節點信息被 RenewDecorate  包裝起來了，這個就是實現註冊信息續約和驅逐的關鍵：

private T               renewal;  // 節點對象封裝
    private long            beginTimestamp; // 註冊事件
    private volatile long   lastUpdateTimestamp; // 續約時間
    private long            duration; // 超時時間

該對象爲註冊節點信息，附加了註冊時間、上次續約時間、過時時間。那麼續約操做就是修改lastUpdateTimestamp，是否過時就是判斷System.currentTimeMillis() - lastUpdateTimestamp > duration 是否成立，成立則認爲節點超時進行驅逐。

和註冊同樣，續約請求的處理 Handler 爲ReNewNodesRequestHandler，最終交由 StoreService 進行續約操做。另一點，續約的時候若是沒有查詢到註冊節點，會觸發節點註冊的操做。

驅出的操做是由定時任務完成，MetaServer 在啓動時會啓動多個定時任務，詳見ExecutorManager#startScheduler,，其中一個任務會調用Registry#evict，其實現爲遍歷存儲的 Map, 得到過時的列表，調用StoreService#removeNodes方法，將他們從 Repository  中移除，這個操做也會觸發變動通知。該任務默認每3秒執行一次。

節點列表變動推送

上文有介紹到，在處理節點註冊請求後，也會存儲一個節點變動事件，即：

dataConfirmStatusService.putConfirmNode(dataNode, DataOperator.ADD);

DataConfirmStatusService  也是一個由 Raft 協議進行同步的存儲，其存儲結構爲：

BlockingQueue<NodeOperator>  expectNodesOrders = new LinkedBlockingQueue();

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

• expectNodesOrders 用來存儲節點變動事件；
• expectNodes 用來存儲變動事件須要確認的節點，也就是說 NodeOperator  只有獲得了其餘節點的確認，纔會從 expectNodesOrders 移除；

那麼事件存儲到 BlockingQueue 裏，哪裏去消費呢？ 看源碼發現，並非想象中的使用一個線程阻塞的讀。

在ExecutorManager中會啓動一個定時任務，輪詢該隊列有沒有數據。即週期性的調用Registry#pushNodeListChange方法，獲取隊列的頭節點並消費。Data 和 Session 各對應一個任務。具體流程以下圖所示：

1. 首先獲取隊列（expectNodesOrders）頭節點，若是爲Null直接返回；
2. 獲取當前數據中心的節點列表，並存儲到確認表（expectNodes）；
3. 提交節點變動推送任務（firePushXxListTask)；
4. 處理任務，即調用 XxNodeService 的 pushXxxNode 方法，即經過 ConnectionHandler 獲取全部的節點鏈接，發送節點列表；
5. 收到回覆後，若是須要確認，則會調用StroeService#confirmNodeStatus 方法，將該節點從expectNodes中移除；
6. 待全部的節點從 expectNodes 中移除，則將這次操做從 expectNodesOrders 移除，處理完畢；

節點列表查詢

Data，Meta,Session Server 都提供 getNodesRequestHandler ，用於處理查詢當前節點列表的請求，其本質上從底層存儲 Repository 讀取數據返回，這裏不在贅述。返回的結果的具體結構見 NodeChangeResult 類，包含各個數據中心的節點列表以及版本號。

基於 Raft 的存儲

後端 Repository 能夠看做SOFAJRaft 的狀態機，任何對 Map 的操做都會在集羣內部，交由 Raft 協議進行同步，從而達到集羣內部的一致。從源碼上看，全部的操做都是直接調用的 RepositoryService 等接口，那麼是如何和 Raft 服務結合起來的呢？

看源碼會發現，凡是引用 RepositoryService 的地方，都加了 @RaftReference， RepositoryService 的具體實現類都加了 @RaftService 註解。其關鍵就在這裏，其處理類爲 RaftAnnotationBeanPostProcessor。具體流程以下：

在 processRaftReference  方法中，凡是加了 @RaftReference 註解的屬性，都會被動態代理類替換，其代理實現見 ProxyHandler 類，即將方法調用，封裝爲 ProcessRequest，經過 RaftClient 發送給 RaftServer。

而被加了 @RaftService 的類會被添加到 Processor 類 中，經過 serviceId(interfaceName + uniqueId) 進行區分。RaftServer 收到請求後，會把它生效到 SOFAJRaft 的狀態機，具體實現類爲 ServiceStateMachine，即會調用 Processor 方法，經過 serviceId 找到這個實現類，執行對應的方法調用。

固然若是本機就是主節點, 對於一些查詢請求不須要走Raft協議而直接調用本地實現方法。 

這個過程其實和 RPC 調用很是相似，在引用方發起的方法調用，並不會真正的執行方法，而是封裝成請求發送到 Raft 服務，由 Raft 狀態機進行真正的方法調用，好比把節點信息存儲到 Map 中。全部節點之間的數據一致由Raft協議進行保證。

總結

在分佈式系統中，集羣成員管理是避不開的問題，有些集羣直接把列表信息寫到配置文件或者配置中心，也有的集羣選擇使用 zookeeper 或者 etcd 等維護集羣元數據，SOFARegistry 選擇基於一致性協議 Raft，開發獨立的MetaServer，來實現集羣列表維護和變動實時推送，以提升集羣管理的靈活性和集羣的健壯性。

SOFARegistryLab 系列閱讀

• 螞蟻金服服務註冊中心 SOFARegistry 解析 | 服務發現優化之路
• 海量數據下的註冊中心 - SOFARegistry 架構介紹