Fabric 中基于Kafka的排序服務(wù)

1. 介紹

我們使用 Kafka 來支持在CFT（crash fault tolerant）場景下的排序服務(wù)以及對多鏈的支持惫东。排序服務(wù)由這些部分組成：

• (1). 一個Kafka 的集群以及它自己對應的ZooKeeper ensemble
• (2). 排序服務(wù)節(jié)點的集合焕盟，這些節(jié)點在排序服務(wù)的客戶端 與 Kafka 集群之間

Fig. 1. An ordering service, consisting of 5 ordering service nodes (OSN-n), and a Kafka cluster. The ordering service client can be connected to multiple OSNs. Note that the OSNs do not communicate with each other directly.

這些排序節(jié)點（ordering service nodes, OSNs）：

• (1). 進行客戶端認證
• (2). 允許客戶端寫一個鏈，或者使用一個簡單的接口讀一個鏈
• (3). 它們也進行 transaction 過濾吮炕，也可以對 configuration transactinos 做驗證擂红，不論是對已有鏈的 reconfig 還是新建一個鏈

2. Where does Kafka come in?(譯者：不知道怎么翻譯)

在 Kafka 中瘪松，Messages (records) 被寫入主題分區(qū) (topic partition)。一個 Kafka 集群可以有多個主題 (topics)悠瞬，每個主題可以有多個分區(qū) (partitions)们豌，如圖2。每個分區(qū)是一個有序的阁危，不可變的記錄序列玛痊，并且被不斷附加。

Fig. 2. A topic that consists of three partitions. Each record in a partition is tagged with an offset number. (Diagram taken from the Kafka website.)

Solution 0. 假設(shè)每一條鏈都有一個獨立的分區(qū)狂打。每當OSNs 完成了客戶端的驗證與對 transaction 的過濾擂煞，它們都能把輸入進來的屬于某個特定鏈的客戶端的 transactions 中繼 (relay) 到該鏈對應的分區(qū)上。然后趴乡，他們可以使用該分區(qū)对省，并返回所有OSNs中常見的 transactions 的有序列表。

Fig. 3. Assume all client transactions here belong to the same chain. OSNs relay incoming client transactions to the same partition in the Kafka cluster. They can then read from that partition and get all the transactions that were posted to the network in an order that will be the same for all OSNs.

結(jié)束了嗎晾捏？接著看

在上面的例子中蒿涎，每一個 client transaction 都是一個獨立的 block；OSN 把每一個 transaction 打包進一個 Block 類型的 message 中惦辛，并對其分配一個值劳秋，該值與 Kafka 集群分配給它的偏移值 (offser number) 相同，并對該 block 簽名胖齐，以便審計 (for auditability)玻淑。任何客戶端發(fā)來的 Deliver RPCs消息，都由在該分區(qū)上設(shè)定的一個 Kafka consumer 來回復呀伙，Kafka consumer 會尋找特定的 block/offset number补履。

Problem 1. 但是我們考慮一下 transaction 被傳入的速率，比如說 1K transactions 每秒剿另。那么 排序服務(wù) (ordering service) 則必須每秒生成 1K個簽名箫锤，并且客戶端在接受側(cè) (receiving side) 也必須每秒驗證 1K個簽名贬蛙。眾所周知，簽名與驗簽是很耗費算力的 (expensive)谚攒，所以這樣做不可行

Solution 1. 所以與其把每一個 transaction 放在單獨的 block 中阳准，我們可以批量的把 transactions 放入 block 中。所以以上面的情況為例五鲫，每 1K個transactions 只需要 排序服務(wù) 做一次簽名溺职，客戶端 做一次驗證岔擂。

Problem 2. 但是如果 transactions 進入的速率不是恒定的呢位喂？假設(shè)排序服務(wù)剛發(fā)出了一批 1K 的transaction，現(xiàn)在在內(nèi)存中有999個transactions乱灵，還在等待下一個 transaction 到來以形成新的一批 (new batch)塑崖，但是一個小時過去了，都沒有新的 transaction 被推送到 排序服務(wù)痛倚。那么先前到達的999個 transactions 的交付時間會有一個不可接受的延時

Solution 2. 我們意識到规婆，我們需要一個批處理計時器 (a batch timer)。當新的一批 transactions 的第一個 transation 進入時蝉稳，設(shè)定計時器抒蚜。接著，無論是我們達到了一個block 所能具有的消息的最大數(shù)量 (batchSize)耘戚，還是當計時器計時器當期 (batchTimeout)嗡髓，我們都可以切分出一個新block (cut a block)。

Problem 3. 然而收津，基于時間的 block 的切分需要調(diào)用在 service 節(jié)點中的協(xié)調(diào) (coordination)饿这。基于消息數(shù)量的 block 的切分是容易的撞秋，比如說长捧，每當你在一個分區(qū)中讀取了2條（你設(shè)定的 batchSize）消息，你就可以切出一個塊吻贿。這個分區(qū)對于所有的OSN來說都是一樣的串结，所以無論你接觸哪個OSN，都可以保證客戶端能夠獲得相同的塊序列舅列〖「睿考慮下面的情況，當你設(shè)定了一個 batchTimeOut = 1s剧蹂， 排序服務(wù)中有兩個 OSN声功。一個 block 剛被切分出來，而且接著又通過 OSN1 進來了 (get posted)一個新的 transaction宠叼，該transaction被推送到到分區(qū)中先巴。OSN2在 t=5s 時讀到該交易其爵，然后設(shè)置了一個將在 t=6s 時到期的計時器。OSN1 在 t=5.6s 時從該分區(qū)讀到了該 transaction伸蚯，設(shè)置了對應的計時器摩渺。接著，第二個 transaction 被推送到 (be posted)到該分區(qū)（譯者：其實我對怎么推送的剂邮？通過不同的OSN來推送是否有差別摇幻？等問題并不明白），OSN2 在 t=6.2s 時讀到了它挥萌，而 OSN1 在 t=6.5s 時讀到了它〈乱觯現(xiàn)在出現(xiàn)了這樣一種情況，OSN1 新切出的 block 包含了這兩個 transaction引瀑，而OSN2切出的 block 只包含第一個 transaction狂芋，于是它們生成了兩個不同的 block，區(qū)塊鏈上出現(xiàn)了分歧憨栽，這是不能接受的帜矾。

Solution 3. 因此，基于時間的 block 的切分屑柔，需要一個明確的協(xié)調(diào)信號屡萤。讓我們假設(shè)每一個 OSN 在切出一個新的 block 時都會給分區(qū)推送一個“到切第 X 塊的時間啦！”的消息（其中 X 是一個整數(shù)掸宛，對應在序列中的下一個 block）死陆，并且 OSN 在從一個分區(qū)讀到這個消息之前，它是不會切出新的 block 的旁涤。注意翔曲，讀取到的消息（以下簡稱為 TTC-X）并不一定是它自己發(fā)出的，如果每個 OSN 都在等待自己的 TTC-X劈愚，那么會再次出現(xiàn)分歧瞳遍。于是，每一個 OSN 在收集到了 batchSize 條消息或者接收到第一個 TTC-X 消息后菌羽，都會切出一個 新的 block掠械。這意味著，后續(xù)所有的具有相同 X 值的 TTC-X都會被忽略注祖。如圖4所示猾蒂。

Fig. 4. OSNs post incoming transactions and TTC-X messages to the partition.

我們現(xiàn)在已經(jīng)找到了切出新 block 的方法，不論是基于 transaction 數(shù)量還是基于時間我們都能保證在所有 OSN 中是晨，block 序列的一致性肚菠。

Problem 4. 和把每個 transaction 放入獨立的 block 情況不同的是，現(xiàn)在 block 的編號不會轉(zhuǎn)換為 Kafka 的該分區(qū)的偏移號罩缴。因此蚊逢，如果排序服務(wù)接收到一個 Deliver 請求层扶，比如說從 block 5 開始返回塊，那么它根本就不知道 consumer 應該去找哪個偏移值（譯者：對這段我理解不是很懂烙荷，是指如果要找到對應的 block镜会，不能根據(jù) 相應的transaction 的 offset 找到嗎？）

Solution 4a. 也許我們可以使用 Block 消息類型中 Metadata 域终抽，并且可以讓 OSN 記下該 block所攜帶的（交易的）偏移值范圍（比如戳表，Block 4的 Metadata 包含內(nèi)容：“offset: 63-64”）。那么如果訪問 block 4 （譯者：原文是 block 9昼伴，但我覺得應該是寫錯了） 的客戶端想要從 block 5 開始獲取流匾旭，則它將通過將起始號設(shè)置為65來發(fā)出一個 Deliver RPC，并且 OSN 可以從偏移值65開始 replay 分區(qū)日志亩码，當 transaction 數(shù)量先達到 batchSize或者先找到一個 TTC-X時季率，切下一個 block。這個方法有兩個問題描沟。第一，我們?nèi)绻云浦禐閰?shù)的話鞭光，就違反了 Deliver API 的協(xié)議吏廉，Deliver 要求以 block 編號為參數(shù)；第二惰许，如果一個客戶端錯過了一堆 block席覆，只是想獲得隨機的 block X（在不擁有 block X-1 的情況下），或者最近生成的 block（通過在 SeekInfo 消息中的 "NEWEST" 參數(shù)）客戶端不知道該通過 Deliver 調(diào)用該傳遞的正確的 offset 值是多少汹买，OSN也不知道佩伤。

Solution 4b. 因此每一個 OSN 都需要對每條鏈維護一個表，用來映射每個 block 到它們所包含的第一個 offset晦毙，如表1所示生巡。注意，這就意味著见妒，如果一個 OSN 的內(nèi)建查詢表不包含所請求的 block 號的話孤荣，它就不能回應（accommodate）一個 Deliver 請求。

Table 1. Example lookup table corresponding to Fig. 4. A block number is mapped to the offset number of the first transaction that it should contain

這個查詢表就使得我們不再需要 block metadata须揣，并且可以讓 OSN 在接收到客戶端在 Deliver 請求后找到正確的 offset盐股。OSN 將請求的 block 編號轉(zhuǎn)換為正確的 offset 值，并設(shè)置了一個 consumer 來查找它〕芸ǎ現(xiàn)在疯汁，我們通過維護一些表，克服了上面所說的問題卵酪。但又出現(xiàn)了兩個問題幌蚊。

Problem 5. 第一秸谢，注意，每當 OSN 接收到一個 Deliver 請求霹肝，它必須從請求的 block 號開始從該分區(qū)檢索所有的消息估蹄，將它們打包成 block 并簽名。這一打包與簽名過程在每一個 replay 請求中都會重復沫换，并且耗費昂貴臭蚁。Problem 6. 第二，因為我們在分區(qū)中有我們需要跳過的冗余的消息讯赏，因此回復一個 Deliver 請求并不像設(shè)置一個 consumer垮兑，尋找請求的 offset，replay所有的記錄那樣簡單漱挎；查詢表會在任意時刻被查詢系枪，交付邏輯開始變得越來越復雜，并且對查詢表的檢查會增加延遲磕谅。我們先不考慮第二個問題私爷，針對第一個問題，為了解決它膊夹，我們需要在創(chuàng)建 blocks 時維持它們（peersist these blocks as we create them）衬浑，當我們 replaying（譯者：不知道這個怎么翻譯合適）時，我們只需要傳輸已維持著的 blocks（transmit the persisted blocks）放刨。

Solution 5a. 為了解決這個問題工秩，假設(shè)我們要全部進入Kafka，我們創(chuàng)建另一個分區(qū) - 讓我們稱之為分區(qū)1进统，這意味著我們目前為止保持引用的分區(qū)是分區(qū)0≈遥現(xiàn)在，每當 OSNs 切出一個 block螟碎，它們都會將其推送 (post) 到分區(qū)1眉菱，并且所有的 Deliver 請求都將由這個分區(qū)提供服務(wù)（served by that partitioin）。因為每個 OSN 都把它們切下來的 block 推送到分區(qū)1抚芦，所以我們能想到分區(qū)1中的 block 的順序并不是鏈中 block 的真正順序倍谜；其中的 block 可能重復，并且由于 block 來自所有的 OSNs叉抡，因而 block 的編號不太可能嚴格遞增 —— 如圖5所示尔崔。

Fig. 5. A possible state of partition 1. Most recent message gets appended to the right. Notice how the block numbers across all OSNs are not in a strictly increasing order (3-3-1-2-2-4). However, the numbers per OSN form a strictly increasing sequence as expected (eg. 1-2 for OSN4).

這意味著 Kafka 分配的 offset 值并不能對應 OSN 分配的（連續(xù)的）block 編號，因此如前面所做的一樣褥民，我們在此處也需要維護一個 block-to-offset 的查詢表季春。如表2所示。

以上解決方案能夠工作消返，但注意到载弄，我們之前所說的問題6依舊沒有解決耘拇；在 OSN 側(cè)的 Deliver 邏輯比較復雜并且 查詢表 在每一次請求都需要被查詢。

Solution 6. 想想會發(fā)現(xiàn)宇攻，造成這個問題的原因是“冗余的消息被推送到了分區(qū)”惫叛。無論是分區(qū)0的 TTC-X 消息（參見圖4中的最后一個消息）還是“Block X”消息 它被發(fā)送到分區(qū)1并且小于或等于較早的消息（圖5中的所有中間消息都小于或等于最左側(cè)的塊3）。那我們?nèi)绾螖[脫這些冗余消息呢逞刷？

Solution 6a. 首先嘉涌，讓我們采取如下的規(guī)則：如果你已經(jīng)從分區(qū)收到與您要發(fā)布的內(nèi)容相同的消息（減去簽名），請中止你自己的推送操作夸浅÷刈睿回到圖5中的例子，如果 OSN1 在形成自己的 block 3 之前已經(jīng)看到了分區(qū)1中的 OSN2 的 Block 3消息帆喇，則它將中止其向分區(qū)1的傳輸警医。（我們在這里描述的所有內(nèi)容都具有分區(qū)0和圖4示例的等效示例。）雖然這樣能夠去掉冗余的消息坯钦，但并不能夠完全消除他們预皇。肯定會有OSNs在同一時間發(fā)布相同的塊的情況葫笼，或者相同的塊正在飛往 Kafka 代理集深啤。即使你加了一些其他操作在里面（比如，在每個節(jié)點要把 block 推送到鏈上時路星，先加一個隨機的延時），你依然會面臨這樣的問題诱桂。

Solution 6b. 如果我們有一個負責推送 block 到分區(qū)1中的 OSN 領(lǐng)導節(jié)點洋丐，這回怎么樣呢？有好幾種選出 leader 的方法挥等；例如友绝，你可以使所有 OSN 在 ZooKeeper 中競爭 znode，或者讓第一個在分區(qū)0中發(fā)布 TTC-X 消息的節(jié)點成為 leader肝劲。另一個有趣的方法是迁客，可以讓所有的 OSNs 屬于同一個 consumer group，這意味著每個 OSN 在一個 topic 中都獲得了自己獨立的一個 partition辞槐。（假設(shè)一個 topic 負載了所有鏈的 partition-0掷漱，另一個 topic 負載了所有鏈的 partition-1）于是，負責將 TTC-X 消息發(fā)送到 partition-0或者負責把 block 推送到某條鏈的 partition-1的 OSN就是這條鏈的 partition 0的擁有者 —— 這是由 Kafka 管理的狀態(tài)榄檬，可以通過 Kafka Metadata API 查詢卜范。（向一個不擁有有該鏈/分區(qū)的 OSN 發(fā)送的 Deliver 請求會被重定向至另外一個合適的 OSN。）

上訴方案可以工作鹿榜，但是如果 leader 發(fā)送 Block X 消息給分區(qū)海雪，然后就宕機了锦爵。這條信息還在去往分區(qū)1的路上，但還沒有到達奥裸。其他 OSNs 意識到 leader 宕機了险掀，因為 leader 不再持有 ZooKeeper 的 znode，于是一個新的 leader 被選取出來湾宙。新選出來的 leader 意識到 Block X 在自己的緩存里但還沒有被推送到分區(qū)1樟氢，于是它把 Block X 推送給分區(qū)1。現(xiàn)在创倔，之前在路上的那個 Block X 到達了分區(qū)1嗡害，那么問題來了，我們又再一次得到了重復的 block畦攘。這一系列事件如圖6所示霸妹。

Fig. 6. Having a leader does not eliminate duplicate messages.

Solution 6c. 日志壓縮能解決這個問題嗎？日志壓縮保證了 Kafka 中保存的是一個分區(qū)中每個 message key 的最新的已知 value知押。情景如圖7所示叹螟。

Fig. 7. Log compaction in Kafka.

所以，如果我們使用了日志壓縮台盯，我們可以完全地從分區(qū)中去除重復的部分罢绽，當然，前提是所有的 Block X 消息都攜帶了相同的 key静盅，而不同的 X 對應的 key 是不同的良价。但是因為日志壓縮存儲的是一個 key 的最新版本，OSNs 可能會因為查詢了過期的查詢表而出錯蒿叠。再看到圖7中的例子明垢，假設(shè)列出的所有 key 都對應了 block，有一個僅接收該分區(qū)中的前兩個消息的OSN具有將 block 1 映射到 offset 0 并且將 block 2 映射到 offset 1的查找表市咽。與此同時痊银，由于分區(qū)已經(jīng)被壓縮（如圖7下半部分所示），因此查找 offset 0（或 1） 將導致錯誤施绎。與日志壓縮問題同樣重要的是溯革，分區(qū)1中的 blocks 并不會升序排序（如圖7下半部分所示，blocks 的順序是 1-3-4-5-2-6）谷醉，所以 Deliver 邏輯還是比較復雜致稀。實際上，考慮到查詢表有過時的風險孤紧，所以日志壓縮的方法不太可行豺裆。

所以這里提供的解決方案都沒有解決問題6。我們來看看有沒有辦法解決這個問題，為此臭猜，讓我們回到第一個問題5 - persisting blocks so that replays are faster（譯者：不知道怎么翻譯才好）躺酒。

Solution 5b. 相比與在 Solution 6a 中使用了另一個 Kafka 分區(qū)，在此蔑歌，我們依舊使用分區(qū)0 —— transaction 和 TTC-X 消息所推送到的分區(qū)羹应，并且每個 OSN 為每個鏈維護一個本地日志。如圖8所示次屠。

Fig. 8. The proposed design. For simplicity we assume all transactions here belong to the same chain, but the design works for multiple chains.

這樣做有如下的好處园匹。

第一，它解決了問題6 —— Solution 6d.：處理一個 Deliver 請求現(xiàn)在只需要從本地賬本里順序讀取劫灶。（不會存在重復裸违，因為 OSN 在寫本地日志時會過濾它們，也沒有查詢表本昏。 OSN 會跟蹤它讀到的最后一個 offset 值供汛，只是為了知道當重新連接時從 Kafka consuming 的地方）

第二，它最大限度地利用了 orderer 代碼庫中的通用組件的使用 - 所有現(xiàn)有實現(xiàn)中的 Deliver 路徑基本上都是相同的

本地賬本為 Deliver 請求提供服務(wù)的一個缺點就是涌穆，會比直接從 Kafka 提供服務(wù)慢怔昨。 但是實際上我們從來沒有從 Kafka 直接提供服務(wù)； OSN上總是做了一些處理宿稀。

具體來說趁舀，如果我們討論一下 replay 請求，替代方案（solution 4b 與 solution 5a）仍然需要在 OSNs 上進行處理（無論是 solution 4b 中的打包與對查詢表的查找祝沸，還是在 solution 5a中對查詢表的查找）矮烹，因此在無賬本解決方案（ledger-less solutions）中我們已經(jīng)付出了一定的代價。

If we are talking about Deliver requests to keep current（“tail -f”等效）罩锐，則解決方案6d與4b相比的額外代價是將 block 存儲到賬本并從其中提供擂送。 在解決方案5a中，塊需要經(jīng)由分區(qū)1進行第二次往返唯欣，因此根據(jù)環(huán)境，這可能甚至比6d差搬味。

總體而言境氢，對傳入的客戶端 transaction 與 TTC-X 消息使用單個分區(qū)（每個鏈）并且將生成的 blocks 存儲在本地賬本（每個鏈）的排序服務(wù)在性能與復雜性之間取得了較好的平衡。