面向流數據的分布式時序同步系統的設計與實現

黃偉健，胡懷湘

面向流數據的分布式時序同步系統的設計與實現

黃偉健，胡懷湘

(華北計算技術研究所，北京 100083)

針對目前分布式流數據管理方案的不足，本文從實際的業務需求出發，結合雷達流數據的單消息體規模大、消息流壓力大以及流量波動明顯的特征，采用主從式的分布式結構，利用Akka異步通信框架，設計并實現了一個面向流數據的分布式時序同步系統。該系統可擴展性強，容錯性好，同時也是后續的分布式流數據實時計算系統和實時存儲系統能夠正常運行的重要前提。本文首先分析了系統使用的關鍵技術，然后結合實際背景設計了系統的整體框架，接著從消息類型、數據結構和主從節點的處理流程三個方面詳細剖析了系統的實現細節，最后通過實驗進一步驗證了系統設計的可行性。

流數據管理；Akka；時序同步；分布式系統

0 引言

近年來，在通信領域、交通領域、金融領域、工業監控領域等領域，出現了一種新的數據類型，這類數據來自傳感器采集信息、交通監控數據、實時交通信息等，通常被稱為流數據（streaming data）[1]，它們是一組順序、大量、快速、連續到達的數據序列，一般情況下，可被視為一個隨時間延續而無限增長的動態數據集合。不同類型的流數據也具有不同的特征。雷達流數據除了具有持續不斷到達、潛在規模無限等一般性的流數據的特征之外，還具有如下特征：（1）數據流有明顯的波峰波谷階段，而且各個階段的界限較為清晰；（2）單個消息體的大小一般為10MB左右，每一批次的數據流包含的消息體個數相對固定；（3）數據流分成多路，數據到達各個數據接入服務器的時間相差不大。在雷達流數據的實際處理場景中，流數據經過產生、預處理、時序同步、緩存、實時計算等過程。其中，時序同步的環節是順利進行實時計算的重要前提。

本文從實際業務需求出發，針對目前流數據管理系統的種種不足[2][3]，結合雷達流數據自身的多個特征，采用主從式分布式架構，利用Akka異步通信框架，設計并實現一個可擴展性強、容錯性好的面向流數據的分布式時序同步系統。

1 關鍵技術

1.1分布式系統架構

目前分布式系統主要有兩種主流的分布式結構：主從結構和P2P對等結構[4]。不同的分布式結構有著不同的特點，在面對不同的場景時設計的難度和系統的性能有著很大的差距，因此在實際應用中需要根據不同的業務需求選擇合適的結構。

主從結構如圖1所示，包含一個主節點和若干個從節點。其中主節點一般扮演著管理者的角色，協調管理各個從節點的信息，并保存系統整體的元數據。主從結構的系統設計較為簡單，有著更強的可控制性，但是主節點的性能往往會影響整個系統的運行，容易成為系統的瓶頸。采用主從結構的系統有很多，典型例子有：Hadoop[5]，Storm，Spark等。

圖1 分布式系統主從結構圖

P2P對等結構如圖2圖所示，該結構的系統中每個節點的地位是對等的，所起到的作用也是類似的。這種結構的系統不存在單點故障的問題，但是每個節點必須向系統中的其他節點廣播自己的信息，使得每個節點都知曉系統的整體狀況，這種設計方案也造成了系統有較大的通信開銷。采用P2P對等結構的系統也很常見，典型例子有：Redis[6]，OceanStore，Past等。

圖2 分布式系統P2P結構圖

1.2并發處理框架Akka

Akka[7]是一個開發庫和運行環境，可以用于構建高并發、分布式、可容錯、事件驅動的基于JVM的應用。

1.2.1 Akka的Actor模型

Actor模型并不是最近才出現的概念，早在上世紀70年代早期，Carl Hewitt就已經提出這一模型，目的是為了解決分布式系統中一系列的編程問題[8]。維基百科這樣定義Actor模型——在計算科學領域，Actor模型是一個并行計算模型，它將Actor作為并行計算的基本元素來對待：為了快速響應一個外部發送來的消息，一個Actor能夠根據既定的規則，自己選擇作出某種決策，例如發送更多的響應消息，或創建更多處理邏輯的Actor，或確定以何種方式去響應接收到的下一個消息。Actor模型是一種分布式系統中的高級抽象方式，也是Akka系統中最核心的概念，它封裝了狀態和行為。另外，Actor之間也可以通過交換消息的方式來相互通信。每個Actor都具有自己的Mailbox來接收其他Actor發送過來的消息。通過Actor的抽象，可以明顯簡化鎖和線程管理的工作，而且可以非常容易地開發出邏輯復雜的并發系統。

1.2.2 Akka Cluster

Akka Cluster提供了一個容錯、去中心化、基于P2P的集群服務，而且不會出現單點故障問題。Akka基于分布式系統的Gossip協議[9]，實現集群廣播服務，而且能夠快速檢測失敗情況。

一個Akka集群由若干個成員節點組成，其中每個成員節點的唯一標識是hostname:port:uid，同時所有成員節點之間是完全解耦合的。一個Akka應用程序是分布式的，它具有一個Actor的集合S，而且每個節點可以集合S的一部分Actor，而非全集S。由于Akka集群是基于Gossip協議實現的，所以如果有一個新的成員節點準備加入到原來的Akka集群中，只需要在集群中任意一個成員節點上執行Join命令，集群中其他節點將會獲知該成員節點加入的消息。

Akka集群中各個成員節點之間的狀態關系，如圖3所示。

圖3 Akka集群節點的狀態關系圖

基于Gossip協議收斂過程得到的確定性結果，Akka集群中任何一個成員節點都有可能成為集群的Leader，而且這個過程是透明的。Leader僅僅扮演一種臨時的角色。因為在各輪Gossip收斂過程中，Leader是可以變化的，它負責管理各個成員節點進入和離開集群。一個成員節點加入后，開始處在joining狀態，當所有其他節點因為廣播消息的原因，都看到了該新成員節點，則Leader會將該節點的狀態修改為up。如果一個節點順利地離開原來的Akka集群，它可以預先將其自身的狀態改為leaving。當Leader看到該節點變化為leaving狀態之后，會將該節點的狀態進一步修改為exiting，然后經過一段時間，當所有其他節點都已經看到該節點狀態變為exiting，則Leader將該節點移出集群，其狀態被修改為removed。如果一個節點處在unreachable狀態，基于Gossip協議的定義，Leader無論通過任何辦法來執行操作，系統都是無法收斂到該節點的，所以當節點處于unreachable狀態，它必須被外界執行某些操作來進行強制性改變，使其狀態變成reachable或者down。如果該節點在離開集群之后，如果想重新加入到原來的Akka集群，必須經過重啟并經過若干個步驟加入到集群。

2 系統設計

本文構建的面向流數據的分布式時序同步系統采用主從結構，不僅能夠很好解決多路流數據同步問題，而且具有良好的可擴展性，同時降低了系統的設計難度。本文設計的系統整體架構如圖4所示。整個系統分為消息控制節點和數據同步節點，其中消息控制節點是系統的主節點，只有一個；數據同步節點是系統的從節點，也就是系統中的數據接入服務器。其中，主節點并不處理到來的流數據，只負責保存各個從節點發送過來的同步信息，以及根據數據目前的到達情況和是否超時來決定向各個從節點下達數據發送的命令。各個從節點負責接收上游的數據，并發送一批數據初次到達服務器以及一批數據完全到達服務器的消息給系統的主節點，并根據主節點返回的命令做進一步的處理。主節點與從節點之間通過消息進行通信交互。同步過后的數據經過復制，成為兩份，一份流向實時計算系統，另外一份流向持久化存儲系統。

圖4 面向流數據的分布式時序同步系統架構圖

3 系統實現

3.1消息類型

本系統使用Akka作為分布式組件之間的通信工具。Akka是一個基于Actor模型的異步消息框架，Actor與Actor之間相互獨立，使用隊列的機制收發消息。本系統可以抽象出4種Actor，這4種Actor分別為：主節點接消息Actor（MasterMsgActor），主節點定時Actor（MasterTimerActor），從節點接數據包Actor（SlaveDataActor），從節點接消息Actor（SlaveMsgActor）。Actor之間的通信是通過Akka封裝的消息來完成，本系統涉及到以下5種消息類型（約定如下：<>中的字段為消息內容，ActorA –> ActorB表示消息是從ActorA發送給ActorB）：

（1）：SlaveDataActor -> MasterMsgActor，從節點接到新一組數據的第一個包，向主節點發送消息，記為FIRST。

（2）：SlaveDataActor ->MasterMsg-Actor，從節點接到新一組數據的最后一個包，向主節點發送消息，記為LAST。

（3）：MasterMsgActor -> SlaveMsg-Actor，主節點向所有從節點下達發送該組數據的SEND命令。

（4）：MasterMsgActor ->MasterTimerActor，主節點接消息進程決定該組數據包休眠指定的SleepTime時間，起到定時效果，記為SLEEP。

（5）：MasterTimerActor -> Master-MsgActor，主節點定時進程到達休眠時間后，向主節點的接消息進程發送該組ID，表示休眠時間已到，記為AWAKE。

3.2數據結構

本文構建的面向流數據的分布式時序同步系統采用典型的主從結構，主節點和從節點分別扮演不同的角色。雷達流數據只在從節點進行緩存，而不需要經過主節點。主節點主要負責從節點之間的協調與控制。系統中的主節點與從節點，分別使用不同的數據結構對數據和消息進行組織和管理。這種設計方案達到了“移動計算，不移動數據”的目的，大大減少了系統的通信開銷，提高了系統的響應速度。

主節點的數據結構為：MasterList（LinkedList< MasterTable>）。MasterList是主節點維護的鏈表，用于記錄所有從節點數據包到達的情況。其中，每一項是自定義的類MasterTable。MasterTable類中包含以下幾項：

表1 MasterTable詳細定義

從節點的數據結構有兩個：

（1）PackageList（LinkedList）。這是從節點維護的鏈表，用于記錄到達數據的包個數以及暫時保存流數據的部分消息體。其中，每一項元素Package是自定義的類。Package中包含以下幾項：

表2 Package詳細定義

（2）DeletedGroupSet（Set）。這是從節點維護的集合，集合中的每個元素都是一個GroupID，代表該批次的數據已經完成同步工作，發送給下游的實時計算系統或者存儲系統，如果有重復的數據到來，將不作任何處理。

3.3處理流程

本文構建的面向流數據的分布式時序同步系統分成主節點和從節點，兩者有著截然不同的處理邏輯。由于系統的整體框架都是基于Akka的消息驅動機制建立的，所以當主節點和從節點接到不同的消息或者數據時，將會啟動不同的處理流程。下面分別從主節點和從節點兩個角度來闡述系統的處理流程。

3.3.1 主節點處理流程

第一種情況，主節點接消息進程接到首次消息FIRST后，處理流程如圖5所示，具體步驟如下：

（1）首先檢查MasterList是否為空。

（2）如果為空，添加GroupID到MasterList，并同時啟動定時進程MasterTimerProc。

（3）否則，遍歷MasterList，如果能找到GroupID，則判斷是否超時。

（4）如果超時，則下達SEND命令給各個從節點，然后刪除對應項。

（5）否則，不執行任何動作。

（6）如果遍歷MasterList之后沒有找到相同的GroupID，則添加GroupID進去，并啟動定時進程MasterTimerProc。

圖5 系統主節點接到FIRST消息

第二種情況，主節點接消息進程接到完成消息LAST后，處理流程如圖6所示，具體步驟如下：

（1）首先遍歷MasterList，查找傳入的GroupID。

（2）找到以后，判斷是否超時。

（3）如果超時，則下達SEND命令給各個從節點，然后刪除對應項。

（4）否則，更新FinishNum，然后判斷FinishNum是否等于從節點的個數。

（5）如果相等，說明各個從節點的該組數據已經全部到達，下達SEND命令給各個從節點，然后刪除對應項。

（6）否則，不執行任何動作。

圖6 系統主節點接到LAST消息

第三種情況，主節點接消息進程接到主節點定時線程的喚醒消息AWAKE后，處理流程如圖7所示，具體步驟如下：

（1）首先遍歷MasterList，查找傳入的GroupID。

（2）如果能夠找到，則下達SEND命令給各個從節點，然后刪除對應項。

（3）否則，向主節點定時進程返回MasterList頭部的GroupID以及剩余定時時間。

第四種情況，主節點定時線程接到主節點接消息進程的定時消息SLEEP后，處理流程如圖8所示，具體步驟如下：

（1）休眠傳入的定時時間。

（2）立即返回傳入的GroupID。

3.3.2 從節點處理流程

從節點不僅接收數據，緩存數據，還根據主節點的反饋信息，執行發放數據到實時計算集群和持久化存儲集群的動作。

圖7 系統主節點接到AWAKE消息

圖8 系統主節點定時線程接到SLEEP消息

第一種情況，從節點接到新到達的數據newData后，處理流程如圖9所示，具體步驟如下：

（1）首先檢查DeletedGroupSet是否為空或者DeletedGroupSet中是否不含有newData的GroupID，記為條件1。

（2）如果滿足條件1，然后判斷PackageList是否為空，記為條件2。

（3）如果滿足條件2，則添加newData到PackageList，并發送FIRST消息給主節點。

（4）然后判斷newData的該組所有個數totalNum是否為1，記為條件3。

（5）如果滿足條件3，則發送LAST消息給主節點。

圖9 系統從節點接到新數據newData

（6）如果不滿足條件2，則遍歷PackageList，查找是否有GroupID等于newData的GroupID，記為條件4。

（7）如果滿足條件4，則在GroupID所在位置插入newData，同時count加一。

（8）然后判斷count是否等于newData的totalNum，記為條件5。

（9）如果滿足條件5，則發送LAST消息給主節點。

（10）如果不滿足條件4，則插入newData到PackageList的末尾，并發送FIRST消息到主節點。

（11）判斷條件3，如果滿足，轉入（5）。

第二種情況，從節點接到主節點返回的SEND命令，處理流程如圖10所示，具體步驟如下：

（1）添加GroupID到DeletedGroupSet，標記該GroupID已經被刪除。

（2）遍歷PackageList，找到GroupID對應的該組所有數據，發送到下游的系統。

（3）刪除該組數據。

圖10 系統從節點接到SEND命令

4 實驗驗證

本實驗采用64位的Ubuntu作為服務器的操作系統，Akka版本為2.4.16。本實驗構建了一個面向流數據的分布式時序同步系統，共使用11臺服務器，其中5臺作為數據發射器，5臺作為與數據發射器一對一的數據接入服務器，同時也是時序同步系統的從節點，另外1臺作為時序同步系統的主節點。實驗主要針對分布式時序同步系統進行功能測試，驗證在不同的流速下，各個從節點服務器之間能否完成同批數據的同步操作。測試方式是在流數據發射器分別發送100 MB/s、200 MB/s、300 MB/s、400 MB/s、500 MB/s的數據包，觀察主節點控制臺顯示的信息。如果每個從節點服務器能夠對到來的每一批流數據成功完成同步操作，主節點將會及時給各個從節點發送命令，將這一批次的數據發送到下游系統，而且該信息將在主節點的控制臺打印出來。如果同步失敗，主節點也將打印同步失敗的信息。

圖11顯示的是當流數據流速為500MB/s時，主節點控制臺打印出來的信息，從圖中顯示的信息可以看出，時序同步系統的運行結果正常，達到了預期的同步效果。其他流速下的結果也類似，系統同樣運行正常，驗證了系統在時序同步方面的正確性。

圖11 流數據流速為500 MB/s時的實驗結果圖

5 結論

本文從實際的業務背景出發，結合雷達流數據的具體特征，借鑒當前分布式系統的設計思想，利用Akka作為異步通信工具，設計并實現了面向流數據的時序同步系統。該系統具有可擴展性強、容錯性好的優點，不僅可以實現雷達流數據的時序同步功能，也可以很好地應用到其他流數據的管理系統中。

[1] BL Golab, MT ?zsu. Issues in data stream management[J] ACM Sigmod Record, 2003,32(2), 4-14.

[2] 馬凱航, 高永明, 吳止鍰, 李磊. 大數據時代數據管理技術研究綜述[J]. 軟件. 2015(10)

[3] 周昭, 林昭文. 基于OpenFlow的數據流管控系統的研究與實現[J]. 軟件. 2013(12)

[4] 華镕. 談談分布式系統[J]. 軟件. 2007(06)

[5] 陸嘉恒. 分布式系統及云計算概論[M]. 北京: 清華大學出版社. 2011.

[6] Redis home page: https://redis.io/

[7] Gupta, Munish. Akka essentials. Packt Publishing Ltd, 2012.

[8] Tasharofi S, Dinges P, Johnson R E. Why do scala developers mix the actor model with other concurrency models? [C]. European Conference on Object-Oriented Programming. Springer Berlin Heidelberg, 2013: 302-326.

[9] Fetahi Wuhib, Rolf Stadler, Mike Spreitzer. A gossip protocol for dynamic resource management in large cloud environments. IEEE transactions on network and service management, 2012.

Design and Implementation of the Distributed Timing Synchronization System for Stream Data

HUANG Wei-jian, HU Huai-xiang
(North China Institute of Computing Technology, Beijing 100083, China)

In view of the shortcomings of the current distributed data management scheme, this paper starts from the actual business demand, combines the characteristics of the radar flow: large single message body, large message flow pressure and obvious fluctuation of the traffic flow, takes the master-based distributed structure and Akka, and finally designs and implements a distributed timing synchronization system. The system is highly scalable and fault-tolerant, and it is also an important prerequisite for the subsequent operation of the distributed real-time computing system. This paper first analyzes the key technology used in the system, and then designs the overall framework of the system, and then analyzes the details of the system from three aspects: the message type, the data structure and the processing flow of the master and subordinate nodes. Finally, And the feasibility of the system is verified by the test.

Stream data management; Akka; Timing synchronization; Distributed system

TP391

: A

10.3969/j.issn.1003-6970.2017.02.022

黃偉健，男，碩士研究生，研究方向為大數據處理及分布式存儲；胡懷湘，男，研究員級高級工程師，研究方向為計算機網絡裝備和網絡存儲技術。

本文著錄格式：黃偉健，胡懷湘. 面向流數據的分布式時序同步系統的設計與實現[J]. 軟件，2017，38（2）：105-111