以下文章來源于波波微課 ，作者架構師楊波

介紹

系統架構微服務化以后，根據微服務獨立數據源的思想，每個微服務一般具有各自獨立的數據源，但是不同微服務之間難免需要通過數據分發來共享一些數據，這個就是微服務的數據分發問題。Netflix/Airbnb等一線互聯網公司的實踐[參考附錄1/2/3]表明，數據一致性分發能力，是構建松散耦合、可擴展和高性能的微服務架構的基礎。

本文解釋分布式微服務中的數據一致性分發問題，應用場景，并給出常見的解決方法。本文主要面向互聯網分布式系統架構師和研發經理。

為啥要分發數據？場景？

我們還是要從具體業務場景出發，為啥要分發數據？有哪些場景？在實際企業中，數據分發的場景其實是非常多的。假設某電商企業有這樣一個訂單服務Order Service，它有一個獨立的數據庫。同時，周邊還有不少系統需要訂單的數據，上圖給出了一些例子：

1. 一個是緩存系統，為了提升訂單數據的訪問性能，我們可以把頻繁訪問的訂單數據，通過redis緩存起來；
2. 第二個是Fulfillment Service，也就是訂單履行系統，它也需要一份訂單數據，借此實現訂單履行的功能；
3. 第三個是ElasticSearch搜索引擎系統，它也需要一份訂單數據，可以支持前臺用戶、或者是后臺運營快速查詢訂單信息；
4. 第四個是傳統數據倉庫系統，它也需要一份訂單數據，支持對訂單數據的分析和挖掘。

當然，為了獲得一份訂單數據，這些系統可以定期去訂單服務查詢最新的數據，也就是拉模式，但是拉模式有兩大問題：

1. 一個是拉數據通常會有延遲，也就是說拉到的數據并不實時；
2. 如果頻繁拉的話，考慮到外圍系統眾多(而且可能還會增加)，勢必會對訂單數據庫的性能造成影響，嚴重時還可能會把訂單數據庫給拉掛。

所以，當企業規模到了一定階段，還是需要考慮數據分發技術，將業務數據同步分發到對數據感興趣的其它服務。除了上面提到的一些數據分發場景，其實還有很多其它場景，例如：

1. 第一個是數據復制(replication)。為了實現高可用，一般要將數據復制多分存儲，這個時候需要采用數據分發。
2. 第二個是支持數據庫的解耦拆分。在單體數據庫解耦拆分的過程中，為了實現不停機拆分，在一段時間內，需要將遺留老數據同步復制到新的數據存儲，這個時候也需要數據分發技術。
3. 第三個是實現CQRS，還有去數據庫Join。這兩個場景我后面有單獨文章解釋，這邊先說明一下，實現CQRS和數據庫去Join的底層技術，其實也是數據分發。
4. 第四個是實現分布式事務。這個場景我后面也有單獨文章講解，這邊先說明一下，解決分布式事務問題的一些方案，底層也是依賴于數據分發技術的。
5. 其它還有流式計算、大數據BI/AI，還有審計日志和歷史數據歸檔等場景，一般都離不開數據分發技術。

總之，波波認為，數據分發，是構建現代大規模分布式系統、微服務架構和異步事件驅動架構的底層基礎技術。

雙寫？

對于數據分發這個問題，乍一看，好像并不復雜，稍有開發經驗的同學會說，我在應用層做一個雙寫不就可以了嗎？比方說，請看上圖右邊，這里有一個微服務A，它需要把數據寫入DB，同時還要把數據寫到MQ，對于這個需求，我在A服務中弄一個雙寫，不就搞定了嗎？其實這個問題并沒有那么簡單，關鍵是你如何才能保證雙寫的事務性？

請看上圖左邊的代碼，這里有一個方法updateDbThenSendMsgInTransaction，這個方法上加了事務性標注，也就是說，如果拋異常的話，數據庫操作會回滾。我們來看這個方法的執行步驟：

第一步先更新數據庫，如果更新成功，那么result設為true，如果更新失敗，那么result設為false；

第二步，如果result為true，也就是說DB更新成功，那么我們就繼續做第三步，向mq發送消息

如果發消息也成功，那么我們的流程就走到第四步，整個雙寫事務就成功了。

如果發消息拋異常，也就是發消息失敗，那么容器會執行該方法的事務性回滾，上面的數據庫更新操作也會回滾。

初看這個雙寫流程沒有問題，可以保證事務性。但是深入研究會發現它其實是有問題的。比方說在第三步，如果發消息拋異常了，并不保證說發消息失敗了，可能只是由于網絡異常抖動而造成的拋異常，實際消息可能是已經發到MQ中，但是拋異常會造成上面數據庫更新操作的回滾，結果造成兩邊數據不一致。

模式一：事務性發件箱(Transactional Outbox)

對于事務性雙寫這個問題，業界沉淀下來比較實踐的做法，其中一種，就是采用所謂事務性發件箱模式，英文叫Transactional Outbox。據說這個模式是eBay最早發明和使用的。事務性發件箱模式不難理解，請看上圖。

我們仍然以訂單Order服務為例。在數據庫中，除了訂單Order表，為了實現事務性雙寫，我們還需增加了一個發件箱Outbox表。Order表和Outbox表都在同一個數據庫中，對它們進行同時更新的話，通過數據庫的事務機制，是可以實現事務性更新的。

下面我們通過例子來展示這個流程，我們這里假定Order Service要添加一個新訂單。

首先第一步，Order Service先將新訂單數據寫入Order表，然后它再向Outbox表中寫入一條訂單新增記錄，這兩個DB操作可以包在一個DB事務里頭，也就是可以實現事務性寫入。

然后第二步，我們再引入一個稱為消息中繼Message Relay的角色，它負責定期Poll拉取Outbox中的新數據，然后第三步再Publish發送到MQ。如果寫入MQ確認成功，Message Relay就可以將Outbox中的對應記錄標記為已消費。這里可能會出現一種異常情況，就是Message Relay在將消息發送到MQ時，發生了網絡抖動，實際消息可能已經寫入MQ，但是Message Relay并沒有得到確認，這時候它會重發，直到明確成功為止。所以，這里也是一個At Least Once，也就是至少交付一次的消費語義，消息可能被重復投遞。因此，MQ之后的消費方要做消息去重或冪等處理。

總之，事務性發件箱模式可以保證，對Order表的修改，然后將對應事件發送到MQ，這兩個動作可以實現事務性，也就是實現數據分發的事務性。

注意，這里的Message Relay角色既可以是一個獨立部署的服務，也可以和Order Service住在一起。生產實踐中，需要考慮Message Relay的高可用部署，還有監控和告警，否則如果Message Relay掛了，消息就發不出來，然后，依賴于消息的各種消費方也將無法正常工作。

Transactional Outbox參考實現 ～ Killbill Common Queue

事務性發件箱的原理簡單，實現起來也不復雜，波波這邊推薦一個生產級的參考實現。這個實現源于一個叫killbill的項目，killbill是美國高朋(GroupOn)公司開源的訂閱計費和支付平臺，這個項目已經有超過8～9年的歷史，在高朋等公司已經有不少落地案例，是一個比較成熟的產品。killbill項目里頭有一些公共庫，單獨放在一個叫killbill-commons的子項目里頭，其中有一個叫killbill common queue，它其實是事務性發件箱的一個生產級實現。上圖有給出這個queue的github鏈接。

Killbill common queue也是一個基于DB實現的分布式的隊列，它上層還包裝了EventBus事件總線機制。killbill common queue的總體設計思路不難理解，請看上圖：

在上圖的左邊，killbill common queue提供發送消息API，并且是支持事務的。比方說圖上的postFromTransaction方法，它可以發送一個BusEvent事件到DB Queue當中，這個方法還接受一個數據庫連接Connection參數，killbill common queue可以保證對事件event的數據庫寫入，和使用同一個Connection的其它數據庫寫入操作，發生在同一個事務中。這個做法其實就是一種事務性發件箱的實現，這里的發件箱存的就是事件event。

除了POST寫入API，killbill common queue還支持類似前面提到的Message Relay的功能，并且是包裝成EeventBus + Handler方式來實現的。開發者只需要實現事件處理器，并且注冊訂閱在EventBus上，就可以接收到DB Queue，也就是發件箱當中的新事件，并進行消費處理。如果事件處理成功，那么EvenbBus會將對應的事件從發件箱中移走；如果事件處理不成功，那么EventBus會負責重試，直到處理成功，或者超過最大重試次數，那么它會將該事件標記為處理失敗，并移到歷史歸檔表中，等待后續人工檢查和干預。這個EventBus的底層，其實有一個Dispatcher派遣線程，它負責定期掃描DB Queue(也就是發件箱)中的新事件，有的話就批量拉取出來，并發送到內部EventBus的隊列中，如果內部隊列滿了，那么Dispather Thread也會暫停拉取新事件。

在killbill common queue的設計中，每個節點上的Dispather線程只負責通過自己這個節點寫入的事件，并且在一個節點上，Dispather線程也只有一個，這樣才能保證消息消費的順序性，并且也不會重復消費。

Reaper機制

killbill common queue，其實是一個基于集中式數據庫實現的分布式隊列，為什么說它是分布式隊列呢？請看上圖，killbill common queue的設計是這樣的，它的每個節點，只負責消費處理從自己這個節點寫入的事件。比方說上圖中有藍色/黃色和綠色3個節點，那么藍色節點，只負責從藍色節點寫入，在數據庫中標記為藍色的事件。同樣，黃色節點，只負責從黃色節點寫入，在數據庫中標記為黃色的事件。綠色節點也是類似。這是一種分布式的設計，如果處理容量不夠，只需按需添加更多節點，就可以實現負載分攤。

這里有個問題，如果其中某個節點掛了，比方說上圖的藍色節點掛了，那么誰來繼續消費數據庫中藍色的，還沒有來得及處理的事件呢？為了解決這個問題，killbill common queue設計了一種稱為reaper收割機的機制。每個節點上都還住了一個收割機線程，它們會定期檢查數據庫，看有沒有長時間無人處理的事件，如果有，就搶占標記為由自己負責。比方說上圖的右邊，最終黃色節點上的收割機線程搶到了原來由藍色節點負責的事件，那么它會把這些事件標記為黃色，也就是由自己來負責。

收割機機制，保證了killbill common queue的高可用性，相當于保證了事務性發件箱中的Message Relay的高可用性。

Killbill PersistentBus表結構

基于killbill common queue的EventBus，也被稱為killbill PersistentBus。上圖給出了它的數據庫表結構，其中bus_events就是用來存放待處理事件的，相當于發件箱，主要的字段包括：

1. event_json，存放json格式的原始數據。
2. creating_owner，記錄創建節點，也就是事件是由哪個節點寫入的。
3. processingowner，記錄處理節點，也就是事件最終是由哪個節點處理的；通常由creatingowner自己處理，但也可能被收割，由其它節點處理。
4. processing_state，當前的處理狀態。
5. error_count，處理錯誤計數，超過一定計數會被標記為處理失敗。

當前處理狀態主要包括6種：

1. AVAILABLE，表示待處理
2. IN_PROCESSING，表示已經被dispatcher線程取走，正在處理中
3. PROCESSED，表示已經處理
4. REMOVED，表示已經被刪除
5. FAILED，表示處理失敗
6. REPEATED，表示被其它節點收割了

除了bus_events待處理事件表，還有一個對應的bus-events-history事件歷史記錄表。不管成功還是失敗，最終，事件會被寫入歷史記錄表進行歸檔，作為事后審計或者人工干預的依據。

上圖下方給出了數據庫表的github鏈接，你可以進一步參考學習。

Killbill PersistentBus處理狀態遷移

上圖給出了killbill PersistentBus的事件處理狀態遷移圖。

1. 剛開始事件處于AVAILABLE待處理狀態；
2. 之后事件被dispatcher線程拉取，進入IN_PROCESSING處理中狀態；
3. 之后，如果事件處理器成功處理了事件，那么事件就進入PROCESSED已經處理狀態；
4. 如果事件處理器處理事件失敗，那么事件的錯誤計數會被增加1，如果錯誤計數還沒有超過最大失敗重試閥值，那么事件就會重新進入AVAILABLE狀態；
5. 如果事件的錯誤數量超過了最大失敗重試閥值，那么事件就會進入FAILED失敗狀態；
6. 如果負責待處理事件的節點掛了，那么到達一定的時間間隔，對應的事件會被收割進入REAPED被收割狀態。

上圖有一個通過API觸發進入的REMOVED移除狀態，這個是給通知隊列用的，用戶可以通過API移除對應的通知消息。順便提一下，除了事件/消息隊列，Killbill queue也是支持通知隊列(或者說延遲消息隊列)的。

模式二：變更數據捕獲(Change Data Capture, CDC)

對于事務性雙寫這個問題，業界沉淀下來比較實踐的做法，其中第二種，就是所謂的變更數據捕獲，英文稱為Change Data Capture，簡稱CDC。

變更數據捕獲的原理也不復雜，它利用了數據庫的事務日志記錄。一般數據庫，對于變更提交操作，都記錄所謂事務日志Transaction Log，也稱為提交日志Commit Log，比方說MySQL支持binlog，Postgres支持Write Ahead log。事務日志可以簡單理解為數據庫本地的一個文件隊列，它記錄了按時間順序發生的對數據庫表的變更提交記錄。

下面我們通過例子來展示這個變更數據捕獲的流程，我們這里假定Order Service要添加一個新訂單。

第一步，Order Service將新訂單記錄寫入Order表，并且提交。因為這是一次表變更操作，所以這次變更會被記錄到數據庫的事務日志當中，其中內容包括發生的變更數據。

第二步，我們還需要引入一個稱為Transaction Log Miner這樣的角色，這個Miner負責訂閱在事務日志隊列上，如果有新的變更記錄，Miner就會捕獲到變更記錄。

然后第三步，Miner會將變更記錄發送到MQ消息隊列。同之前的Message Relay一樣，這里的發送到MQ也是At Least Once語義，消息可能會被重復發送，所以MQ之后的消費者需要做去重或者冪等處理。

總之，CDC技術同樣可以保證，對Order表的修改，然后將對應事件發送到MQ，這兩個動作可以實現事務性，也就是實現數據分發的事務性。

注意，這里的CDC一般是一個獨立部署的服務，生產中需要做好高可用部署，并且做好監控告警。否則如果CDC掛了，消息也就發不出來，然后，依賴于消息的各種消費方也將無法正常工作。

CDC開源項目(企業級)

當前，有幾個比較成熟的企業級的CDC開源項目，我這邊收集了一些，供大家學習參考：

1. 第一個是阿里開源的Canal，目前在github上有超過1.4萬顆星，這個項目在國內用得比較多，之前在拍拍貸的實時數據場景，Canal也有不少成功的應用。Canal主要支持MySQL binlog的增量訂閱和消費。它是基于MySQL的Master/Slave機制，它的Miner角色是通過偽裝成Slave來實現的。這個項目的使用文檔相對比較完善，建議大家一步參考學習。
2. 第二個是Redhat開源的Debezium，目前在github上有超過3.2k星，這個項目在國外用得較多。Debezium主要是在Kafka Connect的基礎上開發的，它不僅支持mysql數據庫，還支持postgres/sqlserver/mongodb等數據庫。
3. 第三個是Zendesk開源的Maxwell，目前在github上有超過2.1k星。Maxwell是一個輕量級的CDC Deamon，主要支持MySQL binlog的變更數據捕獲和處理。
4. 第四個是Airbnb開源的SpinalTap，目前在github上有兩百多顆星。SpinalTap主要支持MySQL binlog的變更捕獲和處理。這個項目的星雖然不多，但是它是在Airbnb SOA服務化過程中，通過實踐落地出來的一個項目，值得參考。

對于上面的這些項目，如果你想生產使用的話，波波推薦的是阿里的Canal，因為這個項目畢竟是國內大廠阿里落地出來，而且在國內已經有不少企業落地案例。其它幾個項目，你也可以參考研究。

學習參考 ～ Eventuate-Tram

既然談到這個CDC，這里有必要提到一個人和一本書，這個人叫Chris Chardson，他是美國的老一輩的技術大牛，曾今是第一代的Cloud Foundry項目的創始人(后來Cloud Foundry被Pivotal所收購)。近幾年，Chris Chardson開始轉戰微服務領域，這兩年，他還專門寫了一本書，叫《微服務設計模式》，英文名是《Microservices Patterns》。這本書主要是講微服務架構和設計模式的，內容還不錯，是我推薦大家閱讀的。

Charis Chardson還專門開發了一個叫Eventuate-Tram的開源項目(這個項目也有商業版)，另外他的微服務書里頭也詳細介紹了這個項目。這個項目可以說是一個大集成框架，它不僅實現了DDD領域驅動開發模式，CQRS命令查詢職責分離模式，事件溯源模式，還實現了Saga事務狀態機模式。當然，這個項目的底層也實現了CDC變更數據捕獲模式。

波波認為，Charis的項目，作為學習研究還是有價值的，但是暫不建議生產級使用，因為他的東西不是一線企業落地出來的，主要是他個人開發的。至于說Charis的項目能否在一線企業落地，還有待時間的進一步檢驗。

Transactional Outbox vs CDC

好的，前面我介紹了解決數據的事務性分發的兩種落地模式，一種是事務性發件箱模式，另外一種是變更數據捕獲模式，這兩種模式其實各有優劣，為了幫助大家做選型決策，我這邊對這兩種模式進行一個比較，請看上面的比較表格：

1. 首先比較一下復雜性，事務性發件箱相對比較簡單，簡單做法只需要在數據庫中增加一個發件箱表，然后再啟一個Poller線程拉消息和發消息就可以了。CDC技術相對比較復雜，需要你深入理解數據庫的事務日志格式和協議。另外Miner的實現也不簡單，要保證不丟消息，如果生產部署的話，還要考慮Miner的高可以部署，還有監控告警等環節。
2. 第二個比較的是Polling延遲和開銷。事務性發件箱的Polling是近實時的，同時如果頻繁拉數據庫表，難免會有性能開銷。CDC是比較實時的，同時它不侵入數據庫和表，所以它的性能開銷相對小。
3. 第三個比較的是應用侵入性。事務性發件箱是有一定的應用侵入性的，應用在更新業務數據的同時，還要單獨發送消息。CDC對應用是無侵入的，因為它拉取的是數據庫事務日志，這個和應用是不直接耦合的。當然，CDC和事務性發件箱模式并不排斥，你可以在應用層采用事務性發件箱模式，同時仍然采用CDC到數據庫去捕獲和發件箱中的消息對應的事務日志。這個方法對應用有一定的侵入性，但是通過CDC可以獲得較好的數據同步性能。
4. 第四點是適用場合。事務性發件箱主要適用于中小規模的企業，因為做法比較簡單，一個開發人員也可以搞定。CDC則主要適用于中大規模互聯網企業，最好有獨立框架團隊負責CDC的治理和維護。像Netflix/Airbnb這樣的一線互聯網公司，也是在中后期才引入CDC技術的[參考附錄1/2/3]。

Single Source of Truth

前面我解答了如何解決微服務的數據一致性分發問題，也給出了可落地的方案。最后，我特別說明在實踐中進行數據分發的一個原則，叫Single Source of Truth，翻成中文就是單一真實數據源。它的意思是說，你要實現數據分發，目標服務可以有很多，但是一定要注意，數據的主人只能有一個，它是數據的權威記錄系統(canonical system of record)，其它的數據都是只讀的，非權威的拷貝(read-only, non-authoritative copy)。

換句話說，任何時候，對于某類數據，它主人應該是唯一的，它是Single Source of Truth，只有它可以修改數據，其它的服務可以獲得數據拷貝，做本地緩存也沒問題，但是這些數據都是只讀的，不能修改。

只有遵循這條原則，數據分發才能正常工作，不會產生不一致的情況。

結論

1. Netflix和Airbnb等一線互聯網公司的實踐證明，企業要真正實現松散耦合、可擴展和高性能的微服務架構，那么底層的數據分發同步能力是非常關鍵的。
2. 數據分發技術，簡單的可以采用事務性發件箱模式來實現，重量級的可以考慮變更數據捕獲CDC技術來實現。事務性發件箱可以參考Killbill Queue的實現，CDC可以參考阿里的Canal等開源產品來實現。
3. 最簡單的雙寫也是實現數據分發的一種方式，但是為了保證一致性，需要引入后臺校驗補償程序。
4. 最后，數據分發/同步的原則是：確保單一真實數據源(Single Source of Truth)。系統中數據的主人應該只有一個，只有主人可以寫入數據，其它都是只讀拷貝。