使用事件和消息隊列實現分佈式事務（轉+補充）

雖然本文並不是筆者原創，可是咱們在非強依賴的事務中原理上也是採用這種方式處理的，不過由於沒有仔細去總結，最近在整理和總結時看到了，故轉載並作部分根據咱們實際狀況的完善和補充。

不一樣於單一架構應用(Monolith), 分佈式環境下, 進行事務操做將變得困難, 由於分佈式環境一般會有多個數據源, 只用本地數據庫事務難以保證多個數據源數據的一致性. 這種狀況下, 可使用兩階段或者三階段提交協議來完成分佈式事務.可是使用這種方式通常來講性能較差, 由於事務管理器須要在多個數據源之間進行屢次等待. 有一種方法一樣能夠解決分佈式事務問題, 而且性能較好, 這就是我這篇文章要介紹的使用事件,本地事務以及消息隊列來實現分佈式事務.

咱們從一個簡單的實例入手. 基本全部互聯網應用都會有用戶註冊的功能. 在這個例子中, 咱們對於用戶註冊有兩步操做: 
1. 註冊成功, 保存用戶信息.
2. 須要給用戶發放一張代金券, 目的是鼓勵用戶進行消費.
若是是一個單一架構應用, 實現這個功能很是簡單: 在一個本地事務裏, 往用戶表插一條記錄, 而且在代金券表裏插一條記錄, 提交事務就完成了. 可是若是咱們的應用是用微服務實現的, 可能用戶和代金券是兩個獨立的服務, 他們有各自的應用和數據庫, 那麼就沒有辦法簡單的使用本地事務來保證操做的原子性了. 如今來看看如何使用事件機制和消息隊列來實現這個需求.(我在這裏使用的消息隊列是kafka, 原理一樣適用於ActiveMQ/RabbitMQ等其餘隊列)

咱們會爲用戶註冊這個操做建立一個事件, 該事件就叫作用戶建立事件(USER_CREATED). 用戶服務成功保存用戶記錄後, 會發送用戶建立事件到消息隊列, 代金券服務會監聽用戶建立事件, 一旦接收到該事件, 代金券服務就會在本身的數據庫中爲該用戶建立一張代金券. 好了, 這些步驟看起來都至關的簡單直觀, 可是怎麼保證事務的原子性呢? 考慮下面這兩個場景:
1. 用戶服務在保存用戶記錄, 還沒來得及向消息隊列發送消息以前就宕機了. 怎麼保證用戶建立事件必定發送到消息隊列了?
2. 代金券服務接收到用戶建立事件, 還沒來得及處理事件就宕機了. 從新啓動以後如何消費以前的用戶建立事件?
這兩個問題的本質是: 如何讓操做數據庫和操做消息隊列這兩個操做成爲一個原子操做. 不考慮2PC, 這裏咱們能夠經過事件表來解決這個問題. 下面是類圖. 

EventPublish是記錄待發布事件的表. 其中:
id: 每一個事件在建立的時候都會生成一個全局惟一ID, 例如UUID.
status: 事件狀態, 枚舉類型. 如今只有兩個狀態: 待發布(NEW), 已發佈(PUBLISHED).
payload: 事件內容. 這裏咱們會將事件內容轉成json存到這個字段裏.
eventType: 事件類型, 枚舉類型. 每一個事件都會有一個類型, 好比咱們以前提到的建立用戶USER_CREATED就是一個事件類型.
EventProcess是用來記錄待處理的事件. 字段與EventPublish基本相同.

咱們首先看看事件的發佈過程. 下面是用戶服務發佈用戶建立事件的順序圖. 
1. 用戶服務在接收到用戶請求後開啓事務, 在用戶表建立一條用戶記錄, 而且在EventPublish表建立一條status爲NEW的記錄, payload記錄的是事件內容, 提交事務.
2. 用戶服務中的定時器首先開啓事務, 而後查詢EventPublish是否有status爲NEW的記錄, 查詢到記錄以後, 拿到payload信息, 將消息發佈到kafka中對應的topic.
發送成功以後, 修改數據庫中EventPublish的status爲PUBLISHED, 提交事務.

下面是代金券服務處理用戶建立事件的順序圖. 
1. 代金券服務接收到kafka傳來的用戶建立事件(其實是代金券服務主動拉取的消息, 先忽略消息隊列的實現), 在EventProcess表建立一條status爲NEW的記錄, payload記錄的是事件內容, 若是保存成功, 向kafka返回接收成功的消息.
2. 代金券服務中的定時器首先開啓事務, 而後查詢EventProcess是否有status爲NEW的記錄, 查詢到記錄以後, 拿到payload信息, 交給事件回調處理器處理, 這裏是直接建立代金券記錄. 處理成功以後修改數據庫中EventProcess的status爲PROCESSED, 最後提交事務.

回過頭來看咱們以前提出的兩個問題:
1. 用戶服務在保存用戶記錄, 還沒來得及向消息隊列發送消息以前就宕機了. 怎麼保證用戶建立事件必定發送到消息隊列了?
根據事件發佈的順序圖, 咱們把建立事件和發佈事件分紅了兩步操做. 若是事件建立成功, 可是在發佈的時候宕機了. 啓動以後定時器會從新對以前沒有發佈成功的事件進行發佈. 若是事件在建立的時候就宕機了, 由於事件建立和業務操做在一個數據庫事務裏, 因此對應的業務操做也失敗了, 數據庫狀態的一致性獲得了保證.
2. 代金券服務接收到用戶建立事件, 還沒來得及處理事件就宕機了. 從新啓動以後如何消費以前的用戶建立事件?
根據事件處理的順序圖, 咱們把接收事件和處理事件分紅了兩步操做. 若是事件接收成功, 可是在處理的時候宕機了. 啓動以後定時器會從新對以前沒有處理成功的事件進行處理. 若是事件在接收的時候就宕機了, kafka會從新將事件發送給對應服務.

經過這種方式, 咱們不用2PC, 也保證了多個數據源之間狀態的最終一致性.
和2PC/3PC這種同步事務處理的方式相比, 這種異步事務處理方式具備異步系統一般都有的優勢:
1. 事務吞吐量大. 由於不須要等待其餘數據源響應.
2. 容錯性好. A服務在發佈事件的時候, B服務甚至能夠不在線.
缺點:
1. 編程與調試較複雜.
2. 容易出現較多的中間狀態. 好比上面的例子, 在用戶服務已經保存了用戶併發布了事件, 可是代金券服務還沒來得及處理以前, 用戶若是登陸系統, 會發現本身是沒有代金券的. 這種狀況可能在有些業務中是可以容忍的, 可是有些業務卻不行. 因此開發以前要考慮好.

另外, 上面的流程在實現的過程當中還有一些能夠改進的地方:
1. 定時器在更新EventPublish狀態爲PUBLISHED的時候, 能夠一次批量更新多個EventProcess的狀態.
2. 定時器查詢EventProcess並交給事件回調處理器處理的時候, 可使用線程池異步處理, 加快EventProcess處理週期.
3. 在保存EventPublish和EventProcess的時候同時保存到Redis, 以後的操做能夠對Redis中的數據進行, 可是要當心處理緩存和數據庫可能狀態不一致問題.
4. 針對Kafka, 由於Kafka的特色是可能重發消息, 因此在接收事件而且保存到EventProcess的時候可能報主鍵衝突的錯誤(由於重複消息id是相同的), 這個時候能夠直接丟棄該消息.

補充點：

一、咱們使用的是rabbitmq cluster；

二、有HA要求，就要防止重複SPOF，而涉及到scheduler的時候，就須要防止重複調度，對此，可使用quartz cluster（能夠容忍必定延時）或者分佈式事務協調器好比zookeeper，前者相對後者系統結構簡單得多，對於這些非強依賴的狀況，由於僅在沒有等待處理的隊列纔會到下一個調度間隔，所以進一步使用多線程以及數據庫優化以後，延時一般在10ms內所有已經push到對方處理了。

三、對於須要回滾和性能要求極高的業務，上述模式沒法直接套用。好比在證券買賣中，延時性是個極爲重要的特性，所以不可能在資金凍結以後最長可能超過10ms才發出買單，對於此，建議的設計是這些功能總體VIP化+可信請求處理。而對於須要回滾的操做，其中一種處理方式是，對於每一個業務功能，都增長一個對應的對衝邏輯，當對端處理失敗的時候，push相應的對衝消息便可。

四、對於某些在短期內發出大量請求的業務邏輯，好比對於一個帳戶，一會兒發出成千上萬請求的業務邏輯，如此一般服務端是經過線程池處理，若是所有請求到到一個線程池中處理，佔據了全部的線程和數據庫鏈接，會致使其餘用戶所有出現hang的狀況，這種狀況下，服務端須要採用pools of single thread executor，而非通用pools的模式，可能還要進一步針對帳號進行二次mod。

在一個系統中，有可能這三種方式都涉及麼，若是確實須要分佈式系統的規模，那麼大部分狀況下會須要兩種結構並存，某些行業和場景中，三種結構都須要。

對於分佈式系統的架構，必定要先按照子系統獨立拆分（這是爲了減小升級、開發、運維複雜性和成本），而後按照業務數據分庫分表拆分（這是爲了儘量晚的引入分佈式事務，分佈式事務一旦引入，極可能使得開發、測試成本劇增，並且在絕大部分狀況下不少系統是拆分過分、架構不合理，而非沒有采用分佈式的緣由），分佈式系統絕對不是、也不能由於數據量、併發數很大就構造分佈式系統，而是由於業務子系統太多、太複雜、又不得不進行大量的子系統間交互才衍生而來。各獨立運行系統間的交互是不能有太多功能的，不然業務架構的設計上就存在疑慮，爲了分佈式而分佈式會使得系統的維護成本極爲高昂，同時分佈式系統大量依賴於MQ和RPC框架，可是不是使用了RPC和MQ就必定要將系統複雜到分佈式系統。