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