取代 ZooKeeper！高併發下的分散式一致性開源元件 StateSynchronizer
StateSynchronizer 是開源分散式流儲存平臺 Pravega 的核心元件。StateSynchronizer 元件以 stream 為基礎，對外提供一致性狀態共享服務。StateSynchronizer 允許一組程序同時讀寫同一共享狀態而不必擔心一致性問題。本文將從共享狀態和一致性的角度出發，詳細描述 StateSynchronizer 的整體架構、工作機制和實現細節。利用 stream 的天然特性，StateSynchronizer 可以高效地確定出更新操作的全域性順序，並且從邏輯上實現了對共享狀態的一致性更新與儲存。由於 stream 訪問的高效與輕量， StateSynchronizer 特別適用於高併發 (>= 10000 clients) 的場景，並在此場景下可以作為替代 ZooKeeper 和 etcd 的解決方案。
StateSynchronizer 設計者之一 Flavio 是著名開源元件 ZooKeeper 的最早作者，他同時也是 《ZooKeeper：分散式過程協同技術詳解》 這本書的作者。
StateSynchronizer 不僅是 Pravega 公共 API 的一部分，許多 Pravega 內部元件也大量依賴 StateSynchronizer 共享狀態，如 ReaderGroup 的元資訊管理。並且我們可以基於 StateSynchronizer 實現更高階的一致性原語，例如跨 stream 的事務。
開源專案地址： https://github.com/pravega/pravega/tree/v0.4.0
1 背景簡介
1.1 什麼是 StateSynchronizer（狀態同步器）
Pravega [1] 既可以被想象成是一組流儲存相關的原語，因為它是實現資料持久化的一種方式，Pravega 也可以被想象成是一個訊息訂閱 - 釋出系統，因為通過使用 reader，writer 和 ReaderGroup 它可以自適應地進行訊息傳遞。本文假設讀者已經熟悉 Pravega 的有關概念，否則可以參考相應的官方文件 [2] 和已釋出的 4 篇專欄文章（見文末連結）。
Pravega 實現了各種不同的構建模組用以實現 stream 相關原語，StateSynchronizer [2] 就是其中之一，目的在於協調分散式的環境中的各個程序^2。從功能上看，StateSynchronizer 為一組程序提供可靠的共享的狀態儲存服務：允許多個客戶端同時讀取和更新同一共享狀態並保證一致性語義，同時提供資料的冗餘和容錯。從實現上看，StateSynchronizer 使用一個 stream 為叢集中執行的多個程序提供了共享狀態的同步機制，這使得構建分散式應用變得更加簡單。使用 StateSynchronizer，多個程序可以同時對同一個共享狀態進行讀取和修改，而不必擔心一致性問題 [3]。
StateSynchronizer 的最大貢獻在於它提供了一種 stream 原生的一致性儲存方案。由於 stream 具有隻允許追加（Append-Only）的特性，這使得大部分現有的儲存服務都無法很好地應用於 stream 儲存的場景。相比於傳統的狀態儲存方案，stream 原生的儲存使得 StateSynchronizer 具有以下優點：
-
與常見的鍵值儲存（Key/Value Store）不同，StateSynchronizer 支援任意抽象的共享狀態，而不僅僅侷限於維護鍵值集合。
-
與常見的資料儲存不同，StateSynchronizer 以增量的方式維護了共享狀態的整個變更歷史，而不僅僅是維護共享狀態的最新快照。這一特性不僅大大減少了網路傳輸開銷，還使得客戶端可以隨時將共享狀態回滾到任意歷史時刻。
-
與常見的狀態儲存不同，StateSynchronizer 的服務端既不儲存共享狀態本身也不負責對共享狀態進行修改，所有共享狀態的儲存和計算都只發生在客戶端本地。這一特性不僅節約了服務端的計算資源，還增加了狀態計算的靈活性，例如：除了基本的 CAS（Compare-And-Swap）語義，還支援高隔離級別的複雜事務^3。
-
與現有的基於樂觀併發控制（Optimistic Concurrent Control, OCC） [4] [5] 的儲存系統不同，StateSynchronizer 可以不依賴多版本控制機制（Multi Version Concurrent Control, MVCC） [6] [7]。這意味著即使在極端高併發的場景下，狀態更新的提交也永遠不會因版本衝突而需要反覆重試。
StateSynchronizer 無意於也不可能在所有場景中替代傳統的分散式鍵值儲存元件，因為它的執行機制大量依賴 stream 的特性。但是，在具有 stream 原生儲存和較強一致性需求的場景下，StateSynchronizer 可能是一種比其它傳統鍵值儲存服務更為高效的選擇。
1.2 “一致性”的不同語義
在不同的上下文環境中，“一致性”一詞往往有著不同的語義 [8] [9]。在分散式儲存和資料高可用（High Availability）相關的語境下，一致性通常指資料副本（Replica）的一致性 [8]：如何保證分佈在不同機器上的資料副本內容不存在衝突，以及如何讓客戶端看起來就像在以原子的方式操作唯一的資料副本，即線性化（Linearizability） [10]。常見的分散式儲存元件往往依賴單一的 Leader（主節點）確定出特定操作的全域性順序，例如：ZooKeeper [11] 和 etcd [12] 都要求所有的寫操作必須由 Leader 轉發給其它資料副本。資料副本的一致性是分散式系統的難點，但卻並不是一致性問題的全部。
脫離資料副本，在應用層的語境下，一致性通常指資料滿足某種約束條件的不變性（Invariant）[13]，即：指的是從應用程式特定的視角出發，保證多個程序無論以怎樣的順序對共享狀態進行修改，共享狀態始終處於一種“正確的狀態”，而這種正確性是由應用程式或業務自身定義的。例如，對於一個交易系統而言，無論同時有多少個交易在進行，所有賬戶的收入與支出總和始終都應該是平衡的；又如，多程序操作（讀 / 寫）一個共享的計數器時，無論各程序以怎樣的順序讀寫計數器，計數器的終值應該始終與所有程序順序依次讀寫計數器所得到的值相同。參考文獻 [8] 將這種一致性歸類為“事務性的一致性（Transactional Consistency）”，而參考文獻 [9] 則將此類一致性簡單稱為“涉及多物件和多操作的一致性”。應用層的資料一致性語義與資料副本的一致性語義完全不同，即使是一個滿足線性化的分散式系統，也需要考慮應用層的資料一致性問題^4。
1.3 StateSynchronizer 與現有的一致性儲存產品
目前常用的分散式鍵值儲存服務，例如 ZooKeeper 和 etcd，都可以看作是一種對共享狀態進行儲存和維護的元件，即所有鍵值所組成的集合構成了當前的共享狀態。在資料副本層面，ZooKeeper 和 etcd 都依賴共識（Consensus）演算法提供一致性保證。ZooKeeper 使用 ZAB（ZooKeeper’s Atomic Broadcast）協議 [14] 在各節點間對寫操作的提交順序達成共識。在廣播階段，ZAB 協議的行為非常類似傳統的兩階段提交協議。etcd 則使用 Raft 協議 [15] 在所有節點上確定出唯一的寫操作序列。與 ZAB 協議不同，Raft 協議每次可以確認出一段一致的提交序列，並且所有的提交動作都是隱式的。在應用層資料層面，ZooKeeper 和 etcd 都使用基於多版本控制機制的樂觀併發控制提供最基礎的一致性保證。一方面，雖然多版本控制機制提供了基本的 CAS 語義，但是在極端的高併發場景下仍因競爭而存在效能問題。另一方面，僅僅依靠多版本控制機制無法提供更加複雜的一致性語義，例如事務。儘管在資料副本層面，ZooKeeper 和 etcd 都提供很強的一致性語義，但對於應用層面的資料一致性卻還有很大的提升空間：ZooKeeper 無法以原子的方式執行一組相關操作，而 etcd 的事務僅支援有限的簡單操作（簡單邏輯判斷，簡單狀態獲取，但不允許對同一個鍵進行多次寫操作）。
在應用層資料層面，ZooKeeper 和 etcd 都使用多版本控制機制提供最基礎的一致性保證。例如，ZooKeeper 的所有寫操作都支援樂觀併發控制：只有當目標節點的當前版本與期望版本相同時，寫操作才允許成功；而 etcd 則更進一步，還支援非常有限的簡單事務操作。一方面，雖然多版本控制機制提供了基本的 CAS 語義，但是在極端的高併發場景下仍因競爭而存在效能問題。另一方面，僅僅依靠多版本控制機制無法提供更加複雜的一致性語義，例如事務。儘管在資料副本層面，ZooKeeper 和 etcd 都提供很強的一致性語義，但對於應用層面的資料一致性卻還有很大的提升空間：ZooKeeper 無法以原子的方式執行一組相關操作，尤其是同時操縱多個鍵；而 etcd 的事務僅支援非常有限的簡單操作（簡單邏輯判斷，簡單狀態獲取，但不允許對同一個鍵進行多次寫操作）。為應用層資料提供比現有的分散式儲存元件更強的一致性語義（複雜事務）和更高的併發度是 StateSynchronizer 的主要目標，尤其是在 stream 原生場景下，因為傳統的以隨機訪問為主的儲存元件很難適配 stream 儲存的順序特性。得益於 stream 的自身特性，StateSynchronizer 可以不依賴樂觀併發控制和 CAS 語義，這意味著不會出現版本衝突也無需重試，從而更加適用於高併發的場景（2.2.4 小節）。在“無條件寫”模式下，StateSynchronizer 的理論更新提交速度等價於 stream 的寫入速度。
與現有的絕大多數儲存服務不同，StateSynchronizer 反轉了傳統的資料儲存模型（2.2.3 小節）：它並不儲存共享狀態本身，轉而儲存所有作用在共享狀態上的更新操作。一方面，這一反轉的資料模型直接抽象出了共享狀態，使得共享狀態不再侷限於簡單的鍵值儲存，而可以推廣到任意需要一致性語義的狀態。另一方面，反轉資料儲存的同時還不可避免地反轉了資料相關的操作，使得原本大量的服務端狀態計算可以直接在客戶端本地完成（2.2.1 小節）。這一特性不僅大大降低了服務端的資源消耗，同時也使得 StateSynchronizer 可以提供更靈活的更新操作和更強一致性語義：複雜事務。在 StateSynchronizer 的框架中，客戶端提交的所有更新操作都是以原子的方式順序執行的，並且所有更新操作的執行都發生在本地。從邏輯上看，每一個更新操作都等價於一個本地事務操作。這也意味著客戶端可以在更新操作中使用複雜的業務邏輯（幾乎是不受限的操作，只要操作本身的作用是確定性的）而無需擔心一致性問題。
2 實現細節
2.1 StateSynchronizer 的本質
圖 1 StateSynchronizer 的整體架構 [3] StateSynchronizer 包括一個嵌入在應用裡的客戶端和一個用於“儲存”共享狀態的 stream。
從整體架構上看，StateSynchronizer 是一個很典型的客戶端 / 伺服器結構（如圖 1 所示）：它包括一個以庫的形式（當前版本僅支援 Java）嵌入在應用中的客戶端，以及伺服器端的一個對應 stream。從概念上看，StateSynchronizer 服務端負責以 stream 的形式“儲存”共享狀態。嚴格說來，stream 儲存的是更新操作而不是共享狀態本身。2.2.3 小節將對此進行更加深入的討論。
StateSynchronizer 客戶端是一個輕量級的元件，它與所有其它的 stream 客戶端（例如 reader 和 writer）並沒有本質上的不同：StateSynchronizer 客戶端使用標準的 stream API 與伺服器端的 stream 互動，並且伺服器端也並不存在任何特定於 StateSynchronizer 的特性或實現。也就是說，StateSynchronizer 客戶端具有其它 stream 客戶端共同的優點，高效。所有 StateSynchronizer 特定的行為都是在客戶端實現的，伺服器端僅僅用於提供 stream 形式的儲存媒介。StateSynchronizer 的客戶端還非常精巧，核心部分的實現不過數百行程式碼 [16]。
2.2 StateSynchronizer 的工作機制
2.2.1 維護本地共享狀態
從概念上說，每一個 StateSynchronizer 都對應一個共享狀態：所有的客戶端都可以併發地對這個共享狀態進行讀寫操作，並且保持一致性。這個共享狀態既可以很簡單（例如，它可以是一個基本的數值變數），也可以很複雜（例如，它也可以是一個任意複雜的資料結構）。但是，如果從物理實現角度上看，根本不存在這樣一個可以被共享訪問的狀態：每一個 StateSynchronizer 的客戶端都只在各自的本地維護著一個“共享”狀態的副本（Copy），除此以外沒有任何地方儲存這個狀態。所有的讀和寫（更新）操作都是直接作用在這個本地共享狀態副本上：讀操作直接返回本地共享狀態副本，而更新操作作用於本地共享狀態並生成新的共享狀態。
為了達到順序一致性 [8]，所有共享狀態必須滿足全序（Total Order）關係 [17]。如果用符號“≺”表示二元 happens-before 語義 [18]，則任意 N 個狀態必須能夠確定出唯一全域性順序，如下：
(1)
注意，happens-before 關係必須滿足傳遞性，反自反性和反對稱性 [19]。
如果讀者閱讀過 StateSynchronizer 介面 [20] 的實現類 StateSynchronizerImpl ，就會發現它有一個名為 currentState 的 StateT 型別的成員，並且 StateT 型別實現了 Revisioned 介面。這就是 StateSynchronizer 所維護的本地共享狀態副本。 Revisioned 介面僅有兩個成員方法： getScopedStreamName() 用於獲取該狀態對應的 stream 的名字， getRevison() 方法用於獲取該狀態對應的 Revision（一個抽象的版本概念，也可以近似等價為 Kafka 的 offset）。而 Revision 介面最終繼承了 Comparable 介面，允許任意兩個 Revision 進行比較，用於保證共享狀態的全序關係。感興趣的讀者可以繼續閱讀 Revision 介面的標準實現類 RevisionImpl 的 compareTo() 方法，就會發現 Revision 的比較實際上是基於 Segment 偏移量進行的。由於 StateSynchronizer 的底層 stream 僅包含一個 segment，基於該 segment 的偏移量天然就是一個全序關係的良定義（well-defined）。
2.2.2 更新操作的抽象模型
StateSynchronizer 上的更新操作的實現是遞迴式的，也可以說是生成式的。StateSynchronizer 的客戶端接受一個更新操作 u n ，將其成功持久化後（細節將在下文討論）應用於當前的本地共享狀態副本 s n ，從而生成新狀態 s n+1 ，如下：
s n+1 = u n (s n ) (2)
從純數學的角度看，這是一個很典型的一階馬爾科夫模型 / 鏈（Markov Model） [21]：如果把 n 看作是離散的時間，那麼 s n 就構成了系統狀態隨時間遷移（Transition）的一個有序序列，並且該系統在任意時間點的狀態 s n+1 只依賴前一時刻的狀態 s n ，並由當前更新 u n 確定，而與任何其它狀態無關。也可以這麼理解，我們假設了狀態 s n 已經包含了所有之前時刻的狀態資訊。這就是所謂的馬爾科夫假設。為了啟動狀態遷移，我們規定系統必須具有一個起始狀態 s 0 ，而更新操作引起了隨後的狀態遷移。
如果從叢集的視角看，有多個 StateSynchronizer 客戶端獨立同時執行並接受更新操作，而每個客戶端本地的共享狀態則分別經歷著基於馬爾科夫模型的狀態遷移。為保證每個 StateSynchronizer 客戶端的本地共享狀態都能夠收斂於相同的最終狀態，首先要求狀態遷移是確定性的（deterministic），也就是說，更新操作 u n 本身必須是確定性的（我們將在 2.3.1 小節深入討論更新操作與確定性問題）。從這個角度看，上述馬爾可夫鏈其實已經退化成一個普通狀態機。其次，所有的 StateSynchronizer 客戶端必須具有相同的起始狀態 s 0 ，並且以相同的順序應用更新 u n 。整個叢集的這種行為模式非常類似經典的複製狀態機（Replicated State Machine）模型 [22]。複製狀態機模型是一個應用廣泛的分散式模型，許多常見的全序廣播 / 原子廣播協議都是基於該模型進行的，如 ZAB 協議和 Raft 協議等。我們有意忽略了著名的 Paxos 協議 [23] [24]，因為原生的 Paxos 協議並非用於解決全序廣播問題，儘管共識演算法與全序廣播之間確實被證明存在等價關係 [25]。複製狀態機模型可以簡單描述如下：
-
在各自獨立的伺服器節點上放置同一狀態機的例項；
-
接受客戶端請求，並轉譯成狀態機的輸入；
-
確定輸入的順序；
-
按已確定的順序在各個狀態機例項上執行輸入；
-
用狀態機的輸出回覆客戶端；
-
監測各個狀態副本或者狀態機輸出可能出現的差異。
複製狀態機最核心也是最困難的部分是如何確定出一個輸入順序，以便讓每個狀態機例項都嚴格按照該順序執行狀態遷移，從而保證一致性。從整體架構上來說，ZAB 協議和 Raft 協議都依賴單一的主節點確定輸入順序：所有的更新操作只能通過主節點進行，因此順序由主節點唯一確定。所不同的是，ZAB 協議通過顯式的類兩階段提交方法保持廣播更新操作的原子性，而 Raft 協議甚至沒有顯式的提交過程，直接依賴計數的方法實現隱式提交。
在 StateSynchronizer 的場景下，狀態機例項即 StateSynchronizer 客戶端，輸入順序即更新操作的應用順序，執行狀態遷移即應用更新操作至本地共享狀態。StateSynchronizer 使用完全不同的方式解決輸入順序的確定問題，使得 StateSynchronizer 不需要依賴任何主節點。從嚴格意義上說，StateSynchronizer 並不負責維護資料副本，但是其本地共享狀態的維護和更新模型都與資料副本有著相似之處。我們將在下文詳細討論 StateSynchronizer 如何確定輸入順序以及和傳統模型的差別。
如果讀者仔細閱讀過 StateSynchronizer 的原始碼，就會發現 StateSynchronizer 介面內定義有一個名為 UpdateGenerator 的函式式介面。 UpdateGenerator 介面本質上是一個二元消費者：它接受兩個引數，其中一個是 StateT 型別的當前共享狀態，另一個是以 List 形式存在在更新操作（ Update 型別）列表，而列表內 的更新操作最終都將被持久化到相應的 stream 上。從概念上看， UpdateGenerator 介面其實就是公式 2 的等價實現。
2.2.3 只儲存更新操作
在傳統的資料庫模型中，資料庫的伺服器端負責維護一個全域性的持久化的共享狀態，即資料庫中所有資料所組成的一個集合。多個獨立的客戶端同時向伺服器端提交更新操作（事務），更新操作作用於共享狀態上引起狀態改變，而客戶端本地不儲存任何狀態。在這個模型中，伺服器端的共享狀態無論從邏輯上看還是從物理上看，它都是共享的（這與 StateSynchronizer 的共享狀態有很大的不同）：因為幾乎所有的資料庫系統都允許多個事務併發執行。從形式化的角度看，所謂“事務 u i 和 u j 是併發的”指的是它們既不滿足 u i ≺ u j 關係，也不滿足 u j ≺ u i 關係，即 u i 的作用對 u j 不完全可見，並且 u j 的作用對 u i 也不完全可見 [13]。可以不是很精確地將併發理解為：u i 和 u j 之間無法確定順序。也可以從直覺上這樣理解：u i 和 u j 的執行，在時間上存在重疊部分。併發直接導致了資料一致性問題。傳統資料庫模型解決併發問題的手段是設定事務的隔離級別 [26]：併發事務在不同的隔離級別下有著不同的可見性。
StateSynchronizer 擯棄了傳統的資料庫模型，從一個完全不同的角度解決併發問題和狀態機輸入順序問題。其核心思想是，StateSynchronizer 的伺服器端只儲存（持久化）了更新操作本身而不是共享狀態，共享狀態由每個客戶端獨立維護，如 2.2.1 小節所述。由於 StateSynchronizer 架構中並不存在物理上的共享狀態，因此不會因為狀態共享而導致競爭，也不會因此產生併發問題。對於每一個 StateSynchronizer 的客戶端而言，所有的更新操作都是順序地作用於本地的共享狀態副本（物理上順序執行），這也不存在併發問題。但是，單憑這一點還不足以保證共享狀態的一致性，除非能夠保證唯一的更新操作應用順序。StateSynchronizer 的伺服器端用單 segment 的 stream 儲存了所有的更新操作：每一個更新操作作為一個 event 被持久化^5。Stream 的最大特性就是隻允許追加：所有的 event 寫入操作只允許在尾部進行（原子操作），並且一個 event 一旦寫入就不允許修改。這一特性不僅使得多個 writer 可以同時進行寫入並且保持一致性，還使得所有 event 的順序得以唯一確定，即每個 event 最終在 Segment 內的相對順序。所以，對於每一個 StateSynchronizer 客戶端來說，都能夠看見一個一致的有序的更新操作檢視。
細心的讀者可能還希望進一步瞭解伺服器端的 stream 是如保持只允許追加的特性和一致性的。與 Kafka 的訊息代理節點（Broker）直接用本地檔案系統儲存 stream 資料的方法不同，Pravega 的訊息代理節點將資料的儲存完全交由一個抽象的儲存層代理，包括資料副本的維護。目前已經支援的具體儲存層實現包括：BookKeeper [27]，HDFS [28]，Extended S3 [29]，NFS [30] 等等。也就是說，資料副本的實現對訊息代理節點來說是完全透明的。具體的 segment 分層儲存設計細節已經超出本文的討論範圍，感興趣的讀者可以自行閱讀 Pravega 的相關文件 [31]。
StateSynchronizer 的這種資料模型其實非常類似 Change Data Capture（CDC） [32] 和 Event Sourcing [33] 的設計模式：不儲存系統狀態，而是通過推導計算得出 [13]。以 stream 形式存在的更新操作其實可以看作是系統狀態的另一種檢視。從這一檢視出發，不僅能夠推匯出系統的最終狀態，還可以得出系統在歷史任意時刻的狀態。
為了讓所有的更新操作本身都能被持久化到 stream 中，StateSynchronizer 要求所有的更新操作都以類的形式實現，封裝好所有所需的狀態並且支援序列化 / 反序列化。這一點從 StateSynchronizer 的介面定義上也可以反映出來：建立一個 StateSynchronizer 例項必須提供兩個 Serializer 介面例項，分別用於對更新操作和起始狀態作序列化 / 反序列化，並且 UpdateGenerator 介面的定義要求所有更新操作必須實現 Update 介面。
2.2.4 更新操作的寫入模式：條件寫與無條件寫
將更新操作本身持久化到相應的 stream 中是 StateSynchronizer 實現更新操作介面的重要步驟之一，因為只有這樣才能使所有的 StateSynchronizer 客戶端都看見一個全域性唯一的更新操作序列。目前，StateSynchronizer 支援以兩種不同的模式將更新操作持久化到 stream 端：條件寫模式（Conditionally Write）與無條件寫模式（Unconditionally Write）。這兩種更新模式分別有各自的適用場景。
圖 2 條件寫示意圖 每個矩形框代表已經持久化到 stream（右側為尾端）中的一個更新操作。實線框為已經累積到當前某個 StateSynchronizer 客戶端本地狀態的更新操作，而虛線框為尚未作用到本地狀態的更新操作，即：其它 StateSynchronizer 客戶端提交但尚未被當前 StateSynchronizer 客戶端拉取的更新操作。兩條豎線分隔符分別對應當前 StateSynchronizer 客戶端所見的 Revision 以及此時真正的最新 Revision。只要存在虛線框所示的更新操作，或者說只要當前 StateSynchronizer 客戶端所見的 Revision 不是最新，那麼條件寫操作就無法成功完成。
在條件寫模式下（參考 StateSynchronizer 介面上 updateState() 方法的實現），當 StateSynchronizer 客戶端嘗試把一個更新操作寫入 stream 內時需要首先檢查當前本地的共享狀態是否是對應 stream 上的最新狀態。如果是，則寫入成功，可以繼續將該更新操作作用於本地的共享狀態並更新為新狀態；如果不是，說明已經有其它的客戶端搶先往 stream 中寫入了其它更新操作，此時本地的共享狀態已經“過期”，本次寫入失敗，如圖 2 所示。對於寫入失敗的情況，StateSynchronizer 會自動嘗試從 stream 拉取所有缺失的更新，並將所有拉取到的更新順序作用於當前本地共享狀態以便將其更新到最新狀態，然後重試條件寫。這一“失敗 - 重試”的過程可能重複多次，直至寫入成功。從概念上看，條件寫表現出的行為與多執行緒程式設計中的 CAS 操作有著諸多相似之處。
如果讀者仔細思考條件寫的實現細節，不難得出如下的結論：檢查狀態是否過期與實際的 stream 寫入動作必須是一個整體的原子操作，否則將出現競爭條件。事實上，檢查狀態是否過期這一動作並不是在客戶端進行的，而是由 stream 的相關介面直接代理的，否則很難與發生在伺服器端的寫入動作合併為一個原子操作。在閱讀過 StateSynchronizer 實現類 StateSynchronizerImpl 的原始碼之後，讀者會發現一個名為 client 的 RevisionedStreamClient 型別成員。 RevisionedStreamClient 是 StateSynchronizer 客戶端用來與後端 stream 互動的唯一入口，所有 stream 的讀寫操作都通過該介面進行，包括條件寫。 RevisionedStreamClient 介面上有一個名為 writeConditionally() 的方法（即條件寫的真正實現），允許在寫入一個 event 的同時指定一個 Revision。正如其名字所暗示的那樣， Revision 介面可以近似理解為 stream 的“版本”：每次成功的寫入操作都會導致對應 stream 的 Revision 發生變化， writeConditionally() 方法甚至還直接返回該 Revision 以方便客戶端用作多版本併發控制。現在繼續討論 writeConditionally() 方法的行為，只有當 stream 的當前的實際 Revision 與指定的 Revision 相同時（即：從上次成功條件寫入到目前為止都沒有其它的成功寫入發生），真正的寫入動作才發生，否則寫入失敗。很明顯，這是一個典型的樂觀併發控制模式。
聰明的讀者甚至還可以從物理實現角度理解 Revision。從 2.2.1 小節的討論中我們知道，Revision 是基於 segment 內的偏移量實現的，而 segment 本質上就是一個無邊界的位元組流。所謂 stream 的“版本”其實就是 stream 當前尾端的偏移量。由於 stream 只允許追加的特性，往指定偏移位置執行寫入操作時，只有當該偏移確實處於尾端時才能成功。圖 2 中所標記的 Revision 既可以看作是當前本地共享狀態所對應的 stream 版本，也可以看作是當前 StateSynchronizer 客戶端所看見的 stream 尾部位置。從這個角度看，stream 的特性和操作得到了統一。
由於條件寫的失敗 - 重試機制，在某些極端場景下（例如更新操作極度頻繁引起的激烈競爭），可能導致較多次數的重試。並且由於條件寫操作目前並未實現公平機制，理論上可能出現某個客戶端“飢餓”的情況。為應對這種場景，StateSynchronizer 還提供了另一種持久化模式：無條件寫模式。在無條件寫模式下（參考 StateSynchronizer 介面上 updateStateUnconditionally() 方法的實現），StateSynchronizer 客戶端往 stream 寫入更新操作時並不會要求比較 Revision，而是無條件地將該更新操作寫入當前 stream 的實際尾端，並且在寫入成功後也不會更新本地的共享狀態。從實現上看，無條件寫模式下的更新動作其實就是一個簡單的 stream 追加動作。在服務和資源正常的情況下，stream 的追加寫入總是能夠成功的。如果呼叫者希望得到更新操作作用後的共享狀態，則還需要手動拉取一次更新（參考 StateSynchronizer 介面上的 fetchUpdates() 方法）。由於更新操作的件寫入動作與拉取動作之間存在時間視窗，在這段時間內可能已經有其它的客戶端繼續寫入新的更新操作。因此，在拉取得到的更新操作序列上，並不能保證之前提交的更新操作是該序列上的最後一個元素。也就是說，在應用該更新操作之前和之後，可能有其它的更新操作已經作用或繼續作用在當前本地共享狀態上。相反，條件寫模式卻總是能保證所提交的更新一定是最後一個作用在當前本地共享狀態上的操作。根據具體應用場景的不同，這可能是個問題，也可能不是。例如，在無條件寫模式下，所有的更新操作現在都變得不可觀測了：假設你執行了一個無條件的更新操作，往一個共享的集合裡面添加了一個元素。現在，哪怕你立刻進行集合遍歷，也不能保證你一定能夠找到剛剛新增的元素，因為可能存在其它客戶端提交的後續更新操作已經將剛剛新增的元素刪除了。這恐怕是一種與直覺相違背的行為表現。總之，與條件寫相比，無條件寫有著優異的併發效能，但是這一切都是有代價的，例如：犧牲了開發者的可理解性。
2.3 其它問題
2.3.1 更新操作與確定性
StateSynchronizer 的更新操作模型（2.2.2 小節）要求所有更新操作的實現必需是確定性的，因為所有的更新操作都會在每一個 StateSynchronizer 客戶端被重放。對於相同的輸入，如果更新操作本身不能夠產生確定性的結果，即使以完全相同的順序在每一個客戶端被執行，也會破壞共享狀態的最終一致性。根據實際業務場景的不同，這一要求可能是一個問題，也可能不是，例如：
-
不可以使用隨機函式。這一看似簡單的要求實際上限制了不少可能性，很多科學計算依賴隨機函式。
-
不可以使用絕大多數的本地狀態，例如：本地時間，本機硬體資訊等。
-
引用任何外部系統的狀態都需要格外小心可能引入的不一致。例如，如果一個外部系統的狀態會隨時間變化，各個客戶端可能看到各不相同的外部狀態，因為同一個更新操作在每個客戶端被執行的時間點是不確定的。
除了保證更新操作的確定性之外，還需要特別注意更新操作的執行是否具有“副作用”，例如：引發全域性狀態或外部系統狀態的改變。如果回答是肯定的，那麼還需要特別注意這些引發狀態改變的動作介面是否具有冪等性 [34]，因為同一個更新操作不僅會在每個客戶端被執行，即使在同一客戶端也可能被執行多次（2.2.4 小節）。
2.3.2 更新操作與更新丟失問題
有人擔心 StateSynchronizer 是否存在丟失更新問題 [6]。丟失更新問題一般在如下場景發生：兩個程序併發地對同一共享變數進行“讀取 - 修改 - 寫入”組合操作。如果這一組合操作不能夠被作為一個原子操作完成，那麼後寫入的狀態有可能覆蓋另一個寫入操作的結果，導致其中一個修改結果（更新）“丟失”。如 2.2.3 小節所述，所有的更新操作都是在 StateSynchronizer 的客戶端本地順序執行的，因此不存在併發修改共享狀態的場景，也不會產生更新丟失問題。
雖然 StateSynchronizer 客戶端保證了以併發安全的方式執行所有更新操作，但是，一個不正確實現的更新操作仍有可能導致更新丟失問題。如果一個應用需要實現“讀取 - 修改 - 寫入”組合操作，唯一正確的做法是將所有的讀取，修改和寫入動作都封裝在同一個更新操作中，即按如下虛擬碼所示實現更新操作 u n ：
u n ：
> 讀取狀態 s n ；
> 執行修改；
> 生成並返回新狀態 s n+1 ；
原始碼 1 用偽程式碼表示的更新操作一般實現
一種常見的錯誤是在更新操作 u n 外部進行“讀取狀態 s n ”和“執行修改”動作，並將新狀態 s n+1 直接封裝進更新操作 u n 。另一種不那麼直觀的錯誤是，儘管將“讀取”，“修改”和“寫入”動作都封裝進了同一個更新操作，但是在進行“讀取狀態 s n ”動作時有意或無意地使用了某種快取機制，即並非每次都從 StateSynchronizer 獲取當前共享狀態 s n 。這兩種錯誤的實現都將導致很嚴重的丟失更新問題。2.2.4 小節的相關討論解釋了其中的原因：由於條件寫操作可能失敗並重試多次，並且每次重試都意味著 StateSynchronizer 客戶端本地的共享狀態已經改變，任何快取或者等價的行為都將導致實際的“執行修改”動作作用在一個已經過期的舊狀態上，從而導致丟失更新問題。
2.3.3 更新操作的順序執行與效能
在每一個 StateSynchronizer 客戶端上，所有的更新操作都是順序執行並作用在本地共享狀態上的，正所謂“解決併發問題最簡單的辦法就是完全消除併發” [13]。有人擔心更新操作的順序執行是否會顯著降低系統性能。從目前已有的研究看，用單執行緒的方式執行所有事務是完全可行的 [35]，並且在很多現有的資料庫實現中已經被採用，例如：VoltDB/H-Store [36]，Redis [37]，Datomic [38] [39] 等。當然，這對事務本身以及資料集都有所要求 [13]，例如：
-
每個事務必須足夠小，並且足夠快。
-
資料集的活躍部分必須足夠小，以便能夠全部載入實體記憶體。否則，頁面的頻繁換入和換出會引起大量的磁碟 IO 操作，導致事務頻繁阻塞。
-
寫操作的吞吐量必須足夠小，以便單 CPU 核心可以有足夠的能力處理。否則，CPU 運算能力將成為瓶頸。
對於一個 StateSynchronizer 應用來說，無論是共享狀態還是更新操作的設計實現，都必須遵循上述要求。
2.3.4 歷史重放與狀態壓縮
每一個 StateSynchronizer 客戶端在進行啟動後的首次更新操作時，都需要從對應的 stream 拉取所有的歷史更新操作，並重放這些操作以便得到當前最新的共享狀態。如果這是一個長時執行的共享狀態，那麼 stream 內此時可能已經累積了相當數量的更新操作。拉取並重放所有這些更新操作可能需要消耗大量的時間與資源，造成首次更新效能低下。為了應對這種場景，StateSynchronizer 還提供了所謂的狀態壓縮機制。狀態壓縮（compact）是一個特殊的 StateSynchronizer 介面方法，它允許將 StateSynchronizer 客戶端的本地共享狀作為一個新的起始狀態，用條件寫模式重新寫入 stream^6，並且使用 stream 的 mark 機制標記該起始狀態的最新位置^7。StateSynchronizer 客戶端每次拉取更新操作時，都會首先嚐試使用 mark 機制定位到最新的起始狀態並忽略所有之前的更新操作，從而避免了長時間的歷史重放。
如果首次更新操作的效能對於應用程式來說非常重要，那麼開發者可以選擇週期性地進行狀態壓縮。那麼首次更新操作所要拉取和應用的更新運算元量則不會多於一個週期內所累積的更新運算元量，這將大大提升首次更新操作的效能。
3 總結
本文主要從狀態共享和一致性的角度出發，詳細描述了 Pravega 的狀態同步元件 StateSynchronizer 的工作機制和實現細節。StateSynchronizer 支援分散式環境下的多程序同時讀寫共享狀態，並提供一致性保證。StateSynchronizer 具有典型的客戶端 / 伺服器架構，但是卻非常輕量和高效，因為伺服器端僅僅用於提供儲存媒介。StateSynchronizer 的核心工作機制可以歸納為兩個關鍵點：維護本地共享狀態和只儲存更新操作本身。StateSynchronizer 利用 stream 的天然特性實現了更新操作的全域性有序。StateSynchronizer 還提供了條件寫和無條件寫兩種更新寫入模式，可以適用於併發度極高的場景。StateSynchronizer 未來的工作可能集中在如何向開發者提供更加便捷易用的程式設計介面，以減輕開發者的負擔。
Pravega 系列文章計劃
Pravega 根據 Apache 2.0 許可證開源， 0.4 版本 已於近日釋出。我們歡迎對流式儲存感興趣的大咖們加入 Pravega 社群，與 Pravega 共同成長。本篇文章為 Pravega 系列第五篇，系列文章如下：
-
高併發下新的分散式一致性解決方案 (StateSynchronizer)
-
Pravega 的僅一次語義及事務支援
-
與 Apache Flink 整合使用
作者簡介
-
蔡超前：華東理工大學計算機應用專業博士研究生，現就職於 Dell EMC，6 年搜尋和分散式系統開發以及架構設計經驗，現從事流相關的設計與研發工作。
-
滕昱：現就職於 Dell EMC 非結構化資料儲存部門 （Unstructured Data Storage）團隊並擔任軟體開發總監。2007 年加入 Dell EMC 以後一直專注於分散式儲存領域。參加並領導了中國研發團隊參與兩代 Dell EMC 物件儲存產品的研發工作並取得商業上成功。從 2017 年開始，兼任 Streaming 儲存和實時計算系統的設計開發與領導工作。
參考文獻
[1] “Pravega,” Dell EMC, [Online]. Available: https://github.com/pravega/pravega .
[2] “Working with Pravega: State Synchronizer,” Dell EMC, [Online]. Available: https://github.com/pravega/pravega/blob/master/documentation/src/docs/state-synchronizer.md .
[3] “Pravega Concepts,” Dell EMC, [Online]. Available: https://github.com/pravega/pravega/blob/master/documentation/src/docs/pravega-concepts.md .
[4] H. T. Kung and J. T. Robinson, “On optimistic methods for concurrency control,” ACM Transactions on Database Systems, vol. 6, no. 2, pp. 213-226, 1981.
[5] P. A. Bernstein and N. Goodman, “Concurrency Control in Distributed Database Systems,” ACM Computing Surveys, vol. 13, no. 2, pp. 185-221, 1981.
[6] “Concurrency Control,” Wikipedia, [Online]. Available: https://en.wikipedia.org/wiki/Concurrency_control .
[7] “Multiversion Concurrency Control,” Wikipedia, [Online]. Available: https://en.wikipedia.org/wiki/Multiversion_concurrency_control .
[8] P. Viotti and M. Vukolić, “Consistency in Non-Transactional Distributed Storage Systems,” ACM Computing Surveys (CSUR), vol. 49, no. 1, 2016.
[9] P. Bailis, A. Davidson, A. Fekete, A. Ghodsi, J. M. Hellerstein and I. Stoica, “Highly available transactions: virtues and limitations,” in Proceedings of the VLDB Endowment, 2013.
[10] M. P. Herlihy and J. M. Wing, “Linearizability: a correctness condition for concurrent objects,” ACM Transactions on Programming Languages and Systems (TOPLAS) , vol. 12, no. 3, pp. 463-492, 1990 .
[11] “Apache ZooKeeper,” [Online]. Available: https://zookeeper.apache.org/ .
[12] “etcd (GitHub Repository),” [Online]. Available: https://github.com/etcd-io/etcd .
[13] M. Kleppmann, Designing Data-Intensive Applications, O’Reilly Media, 2017.
[14] F. P. Junqueira, B. C. Reed and M. Serafini, “Zab: High-performance broadcast for primary-backup systems,” In DSN, pp. 245-256, 2011.
[15] D. Ongaro and J. Ousterhout, “In search of an understandable consensus algorithm,” in Proceedings of the 2014 USENIX conference on USENIX Annual Technical Conference, Philadelphia, 2014.
[16] “StateSynchronizer Related Source Code in Pravega GitHub Repository,” Dell EMC, [Online]. Available: https://github.com/pravega/pravega/tree/master/client/src/main/java/io/pravega/client/state .
[17] M. Hazewinkel, Ed., Encyclopaedia of Mathematics (set), 1 ed., Springer Netherlands, 1994.
[18] L. Lamport, “Time, clocks, and the ordering of events in a distributed system,” Communications of the ACM, vol. 21, no. 7, pp. 558-565, 1978.
[19] “Happened-before,” Wikipedia, [Online]. Available: https://en.wikipedia.org/wiki/Happened-before .
[20] “StateSynchronizer Interface Definition (v0.4),” Dell EMC, [Online]. Available: https://github.com/pravega/pravega/blob/r0.4/client/src/main/java/io/pravega/client/state/StateSynchronizer.java .
[21] P. A. Gagniuc, Markov Chains: From Theory to Implementation and Experimentation, New Jersey: John Wiley & Sons, 2017.
[22] F. B. Schneider, “Implementing fault-tolerant services using the state machine approach: a tutorial,” ACM Computing Surveys, vol. 22, no. 4, pp. 299-319, 1990.
[23] L. Lamport, “The part-time parliament,” ACM Transactions on Computer Systems, vol. 16, no. 2, pp. 133-169, 1998.
[24] L. Lamport, “Paxos Made Simple,” SIGACT News, vol. 32, no. 4, pp. 51-58, 2001.
[25] X. Défago, A. Schiper and P. Urbán, “Total order broadcast and multicast algorithms: Taxonomy and survey,” ACM Computing Surveys, vol. 36, no. 4, pp. 372-421, 2004.
[26] H. Berenson, P. Bernstein, J. Gray, J. Melton, E. O’Neil and P. O’Neil, “A critique of ANSI SQL isolation levels,” in Proceedings of the 1995 ACM SIGMOD international conference on Management of data, San Jose, California, USA, 1995.
[27] “Apache BookKeeper,” [Online]. Available: https://bookkeeper.apache.org/ .
[28] “Apache Hadoop,” [Online]. Available: https://hadoop.apache.org/ .
[29] “Amazon S3,” Amazon, [Online]. Available: https://aws.amazon.com/s3/ .
[30] “NFS version 4.2 (RFC 7862),” [Online]. Available: https://tools.ietf.org/html/rfc7862 .
[31] “Pravega Segment Store Service (v0.4),” Dell EMC, [Online]. Available: https://github.com/pravega/pravega/blob/r0.4/documentation/src/docs/segment-store-service.md .
[32] “Change Data Capture,” Wikipedia, [Online]. Available: https://en.wikipedia.org/wiki/Change_data_capture .
[33] M. Fowler, “Event Sourcing,” 12 12 2005. [Online]. Available: https://martinfowler.com/eaaDev/EventSourcing.html .
[34] “Idempotence,” Wikipedia, [Online]. Available: https://en.wikipedia.org/wiki/Idempotence .
[35] M. Stonebraker, S. Madden and D. J. Abadi, “The End of an Architectural Era (It’s Time for a Complete Rewrite),” in Proceedings of the 33rd international conference on Very large data bases, Vienna, 2007.
[36] R. Kallman, H. Kimura and J. Natkins, “H-Store: A High-Performance, Distributed Main Memory Transaction Processing System,” Proceedings of the VLDB Endowment, vol. 1, no. 2, pp. 1496-1499, 2008.
[37] “Redis,” [Online]. Available: https://redis.io/ .
[38] R. Hickey, “The Architecture of Datomic,” 2 11 2012. [Online]. Available: https://www.infoq.com/articles/Architecture-Datomic .
[39] “Datomic Cloud,” Cognitect, Inc., [Online]. Available: https://www.datomic.com/ .
更多內容，請關注 AI 前線
