1 引言

分布式事務(wù)是企業(yè)集成中的一個(gè)技術(shù)難點(diǎn)音诈，也是每一個(gè)分布式系統(tǒng)架構(gòu)中都會(huì)涉及到的一個(gè)東西伺通，特別是在這幾年越來(lái)越火的微服務(wù)架構(gòu)中必怜，幾乎可以說(shuō)是無(wú)法避免，本文就圍繞單機(jī)事務(wù)鸭丛，分布式事務(wù)以及分布式事務(wù)的處理方式來(lái)展開(kāi)竞穷。

2 事務(wù)

事務(wù)提供一種“要么什么都不做，要么做全套（All or Nothing）”的機(jī)制鳞溉，她有ACID四大特性

• 原子性（Atomicity）：事務(wù)作為一個(gè)整體被執(zhí)行瘾带，包含在其中的對(duì)數(shù)據(jù)庫(kù)的操作要么全部被執(zhí)行，要么都不執(zhí)行熟菲。
• 一致性（Consistency）：事務(wù)應(yīng)確保數(shù)據(jù)庫(kù)的狀態(tài)從一個(gè)一致?tīng)顟B(tài)轉(zhuǎn)變?yōu)榱硪粋€(gè)一致?tīng)顟B(tài)看政。一致?tīng)顟B(tài)是指數(shù)據(jù)庫(kù)中的數(shù)據(jù)應(yīng)滿足完整性約束。除此之外抄罕，一致性還有另外一層語(yǔ)義允蚣，就是事務(wù)的中間狀態(tài)不能被觀察到（這層語(yǔ)義也有說(shuō)應(yīng)該屬于原子性）。
• 隔離性（Isolation）：多個(gè)事務(wù)并發(fā)執(zhí)行時(shí)呆贿，一個(gè)事務(wù)的執(zhí)行不應(yīng)影響其他事務(wù)的執(zhí)行嚷兔，如同只有這一個(gè)操作在被數(shù)據(jù)庫(kù)所執(zhí)行一樣。
• 持久性（Durability）：已被提交的事務(wù)對(duì)數(shù)據(jù)庫(kù)的修改應(yīng)該永久保存在數(shù)據(jù)庫(kù)中做入。在事務(wù)結(jié)束時(shí)冒晰，此操作將不可逆轉(zhuǎn)。

2.1 單機(jī)事務(wù)

以mysql的InnoDB存儲(chǔ)引擎為例母蛛，來(lái)了解單機(jī)事務(wù)是如何保證ACID特性的翩剪。

Image text

事務(wù)的隔離性是通過(guò)數(shù)據(jù)庫(kù)鎖的機(jī)制實(shí)現(xiàn)的，持久性通過(guò)redo log（重做日志）來(lái)實(shí)現(xiàn)彩郊，原子性和一致性通過(guò)Undo log來(lái)實(shí)現(xiàn)前弯。

2.2 分布式事務(wù)

單機(jī)事務(wù)是通過(guò)將操作限制在一個(gè)會(huì)話內(nèi)通過(guò)數(shù)據(jù)庫(kù)本身的鎖以及日志來(lái)實(shí)現(xiàn)ACID，那么分布式環(huán)境下該如何保證ACID特性那?

2.2.1 XA協(xié)議實(shí)現(xiàn)分布式事務(wù)

2.2.1.1 XA描述

X/Open DTP(X/Open Distributed Transaction Processing Reference Model) 是X/Open 這個(gè)組織定義的一套分布式事務(wù)的標(biāo)準(zhǔn)秫逝，也就是了定義了規(guī)范和API接口恕出，由各個(gè)廠商進(jìn)行具體的實(shí)現(xiàn)。
X/Open DTP 定義了三個(gè)組件： AP违帆，TM浙巫，RM

Image text

• AP(Application Program)：也就是應(yīng)用程序，可以理解為使用DTP的程序
• RM(Resource Manager)：資源管理器刷后，這里可以理解為一個(gè)DBMS系統(tǒng)的畴，或者消息服務(wù)器管理系統(tǒng)，應(yīng)用程序通過(guò)資源管理器對(duì)資源進(jìn)行控制尝胆。資源必須實(shí)現(xiàn)XA定義的接口
• TM(Transaction Manager)：事務(wù)管理器丧裁，負(fù)責(zé)協(xié)調(diào)和管理事務(wù)，提供給AP應(yīng)用程序編程接口以及管理資源管理器

其中在DTP定義了以下幾個(gè)概念

• 事務(wù)：一個(gè)事務(wù)是一個(gè)完整的工作單元含衔，由多個(gè)獨(dú)立的計(jì)算任務(wù)組成煎娇，這多個(gè)任務(wù)在邏輯上是原子的
• 全局事務(wù)：對(duì)于一次性操作多個(gè)資源管理器的事務(wù)二庵，就是全局事務(wù)
• 分支事務(wù)：在全局事務(wù)中，某一個(gè)資源管理器有自己獨(dú)立的任務(wù)缓呛，這些任務(wù)的集合作為這個(gè)資源管理器的分支任務(wù)
• 控制線程：用來(lái)表示一個(gè)工作線程催享，主要是關(guān)聯(lián)AP,TM,RM三者的一個(gè)線程，也就是事務(wù)上下文環(huán)境哟绊。簡(jiǎn)單的說(shuō)因妙，就是需要標(biāo)識(shí)一個(gè)全局事務(wù)以及分支事務(wù)的關(guān)系

如果一個(gè)事務(wù)管理器管理著多個(gè)資源管理器，DTP是通過(guò)兩階段提交協(xié)議來(lái)控制全局事務(wù)和分支事務(wù)匿情。

• 第一階段：準(zhǔn)備階段
  事務(wù)管理器通知資源管理器準(zhǔn)備分支事務(wù)兰迫，資源管理器告之事務(wù)管理器準(zhǔn)備結(jié)果
• 第二階段：提交階段
  事務(wù)管理器通知資源管理器提交分支事務(wù)，資源管理器告之事務(wù)管理器結(jié)果

2.2.1.2 XA的ACID特性

• 原子性：XA議使用2PC原子提交協(xié)議來(lái)保證分布式事務(wù)原子性
• 隔離性：XA要求每個(gè)RMs實(shí)現(xiàn)本地的事務(wù)隔離炬称，子事務(wù)的隔離來(lái)保證整個(gè)事務(wù)的隔離汁果。
• 一致性：通過(guò)原子性、隔離性以及自身一致性的實(shí)現(xiàn)來(lái)保證“數(shù)據(jù)庫(kù)從一個(gè)一致?tīng)顟B(tài)轉(zhuǎn)變?yōu)榱硪粋€(gè)一致?tīng)顟B(tài)”玲躯；通過(guò)MVCC來(lái)保證中間狀態(tài)不能被觀察到据德。

2.2.1.3 XA的優(yōu)缺點(diǎn)

• 優(yōu)點(diǎn)：
  對(duì)業(yè)務(wù)無(wú)侵入，對(duì)RM要求高
• 缺點(diǎn)：
  同步阻塞：在二階段提交的過(guò)程中跷车，所有的節(jié)點(diǎn)都在等待其他節(jié)點(diǎn)的響應(yīng)棘利，無(wú)法進(jìn)行其他操作。這種同步阻塞極大的限制了分布式系統(tǒng)的性能朽缴。
  
  
  單點(diǎn)問(wèn)題：協(xié)調(diào)者在整個(gè)二階段提交過(guò)程中很重要善玫，如果協(xié)調(diào)者在提交階段出現(xiàn)問(wèn)題，那么整個(gè)流程將無(wú)法運(yùn)轉(zhuǎn)密强。更重要的是茅郎，其他參與者將會(huì)處于一直鎖定事務(wù)資源的狀態(tài)中，而無(wú)法繼續(xù)完成事務(wù)操作或渤。
  
  
  數(shù)據(jù)不一致：假設(shè)當(dāng)協(xié)調(diào)者向所有的參與者發(fā)送commit請(qǐng)求之后系冗，發(fā)生了局部網(wǎng)絡(luò)異常，或者是協(xié)調(diào)者在尚未發(fā)送完所有 commit請(qǐng)求之前自身發(fā)生了崩潰薪鹦，導(dǎo)致最終只有部分參與者收到了commit請(qǐng)求掌敬。這將導(dǎo)致嚴(yán)重的數(shù)據(jù)不一致問(wèn)題。
  
  
  容錯(cuò)性不好：如果在二階段提交的提交詢問(wèn)階段中池磁，參與者出現(xiàn)故障奔害，導(dǎo)致協(xié)調(diào)者始終無(wú)法獲取到所有參與者的確認(rèn)信息，這時(shí)協(xié)調(diào)者只能依靠其自身的超時(shí)機(jī)制地熄，判斷是否需要中斷事務(wù)舀武。顯然，這種策略過(guò)于保守离斩。換句話說(shuō)银舱，二階段提交協(xié)議沒(méi)有設(shè)計(jì)較為完善的容錯(cuò)機(jī)制，任意一個(gè)節(jié)點(diǎn)是失敗都會(huì)導(dǎo)致整個(gè)事務(wù)的失敗跛梗。

2.2.2 TCC協(xié)議實(shí)現(xiàn)分布式事務(wù)

2.2.2.1 TCC描述

TCC（Try-Confirm-Cancel）分布式事務(wù)模型相對(duì)于 XA 等傳統(tǒng)模型寻馏，其特征在于它不依賴資源管理器（RM）對(duì)分布式事務(wù)的支持，而是通過(guò)對(duì)業(yè)務(wù)邏輯的分解來(lái)實(shí)現(xiàn)分布式事務(wù)核偿。

Image text

• 第一階段：CanCommit

3PC的CanCommit階段其實(shí)和2PC的準(zhǔn)備階段很像诚欠。協(xié)調(diào)者向參與者發(fā)送commit請(qǐng)求，參與者如果可以提交就返回Yes響應(yīng)漾岳，否則返回No響應(yīng)轰绵。
事務(wù)詢問(wèn)：協(xié)調(diào)者向參與者發(fā)送CanCommit請(qǐng)求。詢問(wèn)是否可以執(zhí)行事務(wù)提交操作尼荆。然后開(kāi)始等待參與者的響應(yīng)
響應(yīng)反饋：參與者接到CanCommit請(qǐng)求之后左腔，正常情況下，如果其自身認(rèn)為可以順利執(zhí)行事務(wù)捅儒，則返回Yes響應(yīng)液样，并進(jìn)入預(yù)備狀態(tài)；否則反饋No巧还。

• 第二階段：PreCommit
  協(xié)調(diào)者在得到所有參與者的響應(yīng)之后鞭莽，會(huì)根據(jù)結(jié)果執(zhí)行2種操作：執(zhí)行事務(wù)預(yù)提交，或者中斷事務(wù)

執(zhí)行事務(wù)預(yù)提交

發(fā)送預(yù)提交請(qǐng)求：協(xié)調(diào)者向所有參與者節(jié)點(diǎn)發(fā)出 preCommit 的請(qǐng)求麸祷，并進(jìn)入 prepared 狀態(tài)澎怒。
事務(wù)預(yù)提交：參與者受到 preCommit 請(qǐng)求后，會(huì)執(zhí)行事務(wù)操作阶牍，對(duì)應(yīng) 2PC 準(zhǔn)備階段中的 “執(zhí)行事務(wù)”喷面，也會(huì) Undo 和 Redo 信息記錄到事務(wù)日志中。
各參與者響應(yīng)反饋:如果參與者成功執(zhí)行了事務(wù)荸恕，就反饋 ACK 響應(yīng)乖酬，同時(shí)等待指令：提交（commit） 或終止（abort）

中斷事務(wù)
發(fā)送中斷請(qǐng)求：協(xié)調(diào)者向所有參與者節(jié)點(diǎn)發(fā)出 abort 請(qǐng)求 。
中斷事務(wù)：參與者如果收到 abort 請(qǐng)求或者超時(shí)了融求，都會(huì)中斷事務(wù)咬像。

• 第三階段：Do Commit
  該階段進(jìn)行真正的事務(wù)提交，也可以分為以下兩種情況

執(zhí)行提交
發(fā)送提交請(qǐng)求：協(xié)調(diào)者接收到各參與者發(fā)送的ACK響應(yīng)生宛，那么他將從預(yù)提交狀態(tài)進(jìn)入到提交狀態(tài)县昂。并向所有參與者發(fā)送 doCommit 請(qǐng)求。
事務(wù)提交：參與者接收到 doCommit 請(qǐng)求之后陷舅，執(zhí)行正式的事務(wù)提交倒彰。并在完成事務(wù)提交之后釋放所有事務(wù)資源。
響應(yīng)反饋：事務(wù)提交完之后莱睁，向協(xié)調(diào)者發(fā)送 ACK 響應(yīng)待讳。
完成事務(wù)：協(xié)調(diào)者接收到所有參與者的 ACK 響應(yīng)之后芒澜，完成事務(wù)。

中斷事務(wù)
協(xié)調(diào)者沒(méi)有接收到參與者發(fā)送的 ACK 響應(yīng)（可能是接受者發(fā)送的不是ACK響應(yīng)创淡，也可能響應(yīng)超時(shí)）痴晦，那么就會(huì)執(zhí)行中斷事務(wù)。
發(fā)送中斷請(qǐng)求：協(xié)調(diào)者向所有參與者發(fā)送 abort 請(qǐng)求琳彩。
事務(wù)回滾：參與者接收到 abort 請(qǐng)求之后誊酌，利用其在階段二記錄的 undo 信息來(lái)執(zhí)行事務(wù)的回滾操作，并在完成回滾之后釋放所有的事務(wù)資源露乏。
反饋結(jié)果：參與者完成事務(wù)回滾之后碧浊，向協(xié)調(diào)者發(fā)送 ACK 消息。
中斷事務(wù)：協(xié)調(diào)者接收到參與者反饋的 ACK 消息之后瘟仿，完成事務(wù)的中斷箱锐。

2.2.2.2 TCC的ACID特性

• 原子性：TCC 模型也使用 2PC 原子提交協(xié)議來(lái)保證事務(wù)原子性。Try 操作對(duì)應(yīng)2PC 的一階段準(zhǔn)備（Prepare）猾骡；Confirm 對(duì)應(yīng) 2PC 的二階段提交（Commit）瑞躺，Cancel 對(duì)應(yīng) 2PC 的二階段回滾（Rollback），可以說(shuō) TCC 就是應(yīng)用層的 2PC兴想。
• 隔離性：隔離的本質(zhì)是控制并發(fā)幢哨，放棄在數(shù)據(jù)庫(kù)層面加鎖通過(guò)在業(yè)務(wù)層面加鎖來(lái)實(shí)現(xiàn)∩┍悖【比如在賬戶管理模塊設(shè)計(jì)中捞镰，增加可用余額和凍結(jié)金額的設(shè)置】
• 一致性：通過(guò)原子性保證事務(wù)的原子提交、業(yè)務(wù)隔離性控制事務(wù)的并發(fā)訪問(wèn)毙替，實(shí)現(xiàn)分布式事務(wù)的一致性狀態(tài)轉(zhuǎn)變岸售；事務(wù)的中間狀態(tài)不能被觀察到這點(diǎn)并不保證[本協(xié)議是基于柔性事務(wù)理論提出的]。

2.2.2.3 TCC的優(yōu)缺點(diǎn)

• 優(yōu)點(diǎn)：
  相對(duì)于二階段提交厂画，三階段提交主要解決的單點(diǎn)故障問(wèn)題凸丸，并減少了阻塞的時(shí)間。因?yàn)橐坏﹨⑴c者無(wú)法及時(shí)收到來(lái)自協(xié)調(diào)者的信息之后袱院，他會(huì)默認(rèn)執(zhí)行 commit屎慢。而不會(huì)一直持有事務(wù)資源并處于阻塞狀態(tài)。
• 缺點(diǎn)：
  三階段提交也會(huì)導(dǎo)致數(shù)據(jù)一致性問(wèn)題忽洛。由于網(wǎng)絡(luò)原因腻惠，協(xié)調(diào)者發(fā)送的 Cancel 響應(yīng)沒(méi)有及時(shí)被參與者接收到，那么參與者在等待超時(shí)之后執(zhí)行了 commit 操作欲虚。這樣就和其他接到 Cancel 命令并執(zhí)行回滾的參與者之間存在數(shù)據(jù)不一致的情況集灌。

2.2.3 SAGA協(xié)議實(shí)現(xiàn)分布式事務(wù)

2.2.3.1 SAGA協(xié)議介紹

Saga的組成：

• 每個(gè)Saga由一系列sub-transaction Ti 組成
• 每個(gè)Ti 都有對(duì)應(yīng)的補(bǔ)償動(dòng)作Ci，補(bǔ)償動(dòng)作用于撤銷(xiāo)Ti造成的結(jié)果

saga的執(zhí)行順序有兩種：

• T1, T2, T3, ..., Tn
• T1, T2, ..., Tj, Cj,..., C2, C1复哆，其中0 < j < n

Saga定義了兩種恢復(fù)策略：

• backward recovery欣喧，向后恢復(fù)腌零，即上面提到的第二種執(zhí)行順序，其中j是發(fā)生錯(cuò)誤的sub-transaction续誉，這種做法的效果是撤銷(xiāo)掉之前所有成功的sub-transation莱没，使得整個(gè)Saga的執(zhí)行結(jié)果撤銷(xiāo)。
• forward recovery酷鸦，向前恢復(fù)，適用于必須要成功的場(chǎng)景牙咏，執(zhí)行順序是類(lèi)似于這樣的：T1, T2, ..., Tj(失敗), Tj(重試),..., Tn臼隔，其中j是發(fā)生錯(cuò)誤的sub-transaction。該情況下不需要Ci妄壶。

Saga的注意事項(xiàng)

• Ti和Ci是冪等的摔握。舉個(gè)例子，假設(shè)在執(zhí)行Ti的時(shí)候超時(shí)了丁寄，此時(shí)我們是不知道執(zhí)行結(jié)果的氨淌，如果采用forward recovery策略就會(huì)再次發(fā)送Ti，那么就有可能出現(xiàn)Ti被執(zhí)行了兩次伊磺，所以要求Ti冪等盛正。如果采用backward recovery策略就會(huì)發(fā)送Ci，而如果Ci也超時(shí)了屑埋，就會(huì)嘗試再次發(fā)送Ci豪筝，那么就有可能出現(xiàn)Ci被執(zhí)行兩次，所以要求Ci冪等摘能。
• Ci必須是能夠成功的续崖，如果無(wú)法成功則需要人工介入。如果Ci不能執(zhí)行成功就意味著整個(gè)Saga無(wú)法完全撤銷(xiāo)团搞，這個(gè)是不允許的严望。但總會(huì)出現(xiàn)一些特殊情況比如Ci的代碼有bug、服務(wù)長(zhǎng)時(shí)間崩潰等逻恐，這個(gè)時(shí)候就需要人工介入了
• Ti - Ci和Ci - Ti的執(zhí)行結(jié)果必須是一樣的：sub-transaction被撤銷(xiāo)了像吻。舉例說(shuō)明，還是考慮Ti執(zhí)行超時(shí)的場(chǎng)景梢莽，我們采用了backward recovery萧豆，發(fā)送一個(gè)Ci，那么就會(huì)有三種情況：
  1：Ti的請(qǐng)求丟失了昏名，服務(wù)之前沒(méi)有涮雷、之后也不會(huì)執(zhí)行Ti
  2：Ti在Ci之前執(zhí)行
  3：Ci在Ti之前執(zhí)行
  對(duì)于第1種情況，容易處理轻局。對(duì)于第2洪鸭、3種情況样刷，則要求Ti和Ci是可交換的（commutative)，并且其最終結(jié)果都是sub-transaction被撤銷(xiāo)览爵。

Saga架構(gòu)

Image text

• Saga Execution Component解析請(qǐng)求JSON并構(gòu)建請(qǐng)求圖
• TaskRunner 用任務(wù)隊(duì)列確保請(qǐng)求的執(zhí)行順序
• TaskConsumer 處理Saga任務(wù)置鼻，將事件寫(xiě)入saga log，并將請(qǐng)求發(fā)送到遠(yuǎn)程服務(wù)

2.2.3.2 SAGA的ACID特性

• 原子性：通過(guò)SAGA協(xié)調(diào)器實(shí)現(xiàn)
• 一致性：本地事務(wù)+SAGA Log
• 持久性：SAGA Log
• 隔離性：不保證（同TCC）

3 分布式事務(wù)的處理方案

3.1 XA

僅在同一個(gè)事務(wù)上下文中需要協(xié)調(diào)多種資源（即數(shù)據(jù)庫(kù)蜓竹，以及消息主題或隊(duì)列）時(shí)箕母，才有必要使用 X/Open XA 接口。數(shù)據(jù)庫(kù)接入XA需要使用XA版的數(shù)據(jù)庫(kù)驅(qū)動(dòng)俱济，消息隊(duì)列要實(shí)現(xiàn)XA需要實(shí)現(xiàn)javax.transaction.xa.XAResource接口嘶是。

3.1.1 jotm的分布式事務(wù)

代碼如下：

public class UserService {
    @Autowired
    private UserDao userDao;
    @Autowired
    private LogDao logDao;
    @Transactional
    public void save(User user){
        userDao.save(user);
        logDao.save(user);
        throw new RuntimeException();
    }
}
@Resource
public class UserDao {
    @Resource(name="jdbcTemplateA")
    private JdbcTemplate jdbcTemplate;
    public void save(User user){
        jdbcTemplate.update("insert into user(name,age) values(?,?)",user.getName(),user.getAge());
    }
}
@Repository
public class LogDao {
    @Resource(name="jdbcTemplateB")
    private JdbcTemplate jdbcTemplate;
    public void save(User user){
        jdbcTemplate.update("insert into log(name,age) values(?,?)",user.getName(),user.getAge());
    }
}

配置：

    <bean id="jotm" class="org.objectweb.jotm.Current" />
    <bean id="transactionManager" class="org.springframework.transaction.jta.JtaTransactionManager">
        <property name="userTransaction" ref="jotm" />
    </bean>
    <tx:annotation-driven transaction-manager="transactionManager"/>
    <!-- 配置數(shù)據(jù)源 -->
    <bean id="dataSourceA" class="org.enhydra.jdbc.pool.StandardXAPoolDataSource"  destroy-method="shutdown">
        <property name="dataSource">
            <bean class="org.enhydra.jdbc.standard.StandardXADataSource" destroy-method="shutdown">
                <property name="transactionManager" ref="jotm" />
                <property name="driverName" value="com.mysql.jdbc.Driver" />
                <property name="url" value="jdbc:mysql://localhost:3306/test?useUnicode=true&amp;characterEncoding=utf-8" />
            </bean>
        </property>
        <property name="user" value="xxx" />
        <property name="password" value="xxx" />
    </bean>
    <!-- 配置數(shù)據(jù)源 -->
    <bean id="dataSourceB"   class="org.enhydra.jdbc.pool.StandardXAPoolDataSource"  destroy-method="shutdown">
        <property name="dataSource">
            <bean class="org.enhydra.jdbc.standard.StandardXADataSource" destroy-method="shutdown">
                <property name="transactionManager" ref="jotm" />
                <property name="driverName" value="com.mysql.jdbc.Driver" />
                <property name="url" value="jdbc:mysql://localhost:3306/test2?useUnicode=true&amp;characterEncoding=utf-8" />
            </bean>
        </property>
        <property name="user" value="xxx" />
        <property name="password" value="xxx" />
    </bean>
    <bean id="jdbcTemplateA" class="org.springframework.jdbc.core.JdbcTemplate">
        <property name="dataSource" ref="dataSourceA" />
    </bean>
    <bean id="jdbcTemplateB" class="org.springframework.jdbc.core.JdbcTemplate">
        <property name="dataSource" ref="dataSourceB" />
    </bean>

使用到的JAR包：

  compile 'org.ow2.jotm:jotm-core:2.3.1-M1'
  compile 'org.ow2.jotm:jotm-datasource:2.3.1-M1'
  compile 'com.experlog:xapool:1.5.0'

事務(wù)配置:
我們知道分布式事務(wù)中需要一個(gè)事務(wù)管理器即接口javax.transaction.TransactionManager、面向開(kāi)發(fā)人員的javax.transaction.UserTransaction蛛碌。對(duì)于jotm來(lái)說(shuō)聂喇，他們的實(shí)現(xiàn)類(lèi)都是Current

public class Current implements UserTransaction, TransactionManager

我們?nèi)绻胧褂梅植际绞聞?wù)的同時(shí)，又想使用Spring帶給我們的@Transactional便利蔚携，就需要配置一個(gè)JtaTransactionManager希太，而該JtaTransactionManager是需要一個(gè)userTransaction實(shí)例的，所以用到了上面的Current酝蜒，如下配置:

<bean id="jotm" class="org.objectweb.jotm.Current" />
<bean id="transactionManager" class="org.springframework.transaction.jta.JtaTransactionManager">  
    <property name="userTransaction" ref="jotm" />  
</bean>
<tx:annotation-driven transaction-manager="transactionManager"/>

同時(shí)上述StandardXADataSource是需要一個(gè)TransactionManager實(shí)例的誊辉，所以上述StandardXADataSource配置把jotm加了進(jìn)去.

執(zhí)行過(guò)程：

• 第一步：事務(wù)攔截器開(kāi)啟事務(wù)
  我們知道加入了@Transactional注解，同時(shí)開(kāi)啟tx:annotation-driven秕硝，會(huì)對(duì)本對(duì)象進(jìn)行代理芥映，加入事務(wù)攔截器。在事務(wù)攔截器中远豺，獲取javax.transaction.UserTransaction奈偏，這里即org.objectweb.jotm.Current，然后使用它開(kāi)啟事務(wù)躯护，并和當(dāng)前線程進(jìn)行綁定惊来，綁定關(guān)系數(shù)據(jù)存放在org.objectweb.jotm.Current中。
• 第二步：使用jdbcTemplate進(jìn)行業(yè)務(wù)操作
  dbcTemplateA要從dataSourceA中獲取Connection,和當(dāng)前線程進(jìn)行綁定棺滞，同時(shí)以對(duì)應(yīng)的dataSourceA作為key裁蚁。同時(shí)判斷當(dāng)前線程是否含有事務(wù)，通過(guò)dataSourceA中的org.objectweb.jotm.Current發(fā)現(xiàn)當(dāng)前線程有事務(wù)继准，則把Connection自動(dòng)提交設(shè)置為false,同時(shí)將該連接納入當(dāng)前事務(wù)中枉证。
  jdbcTemplateB要從dataSourceB中獲取Connection,和當(dāng)前線程進(jìn)行綁定，同時(shí)以對(duì)應(yīng)的dataSourceB作為key移必。同時(shí)判斷當(dāng)前線程是否含有事務(wù)室谚，通過(guò)dataSourceB中的org.objectweb.jotm.Current發(fā)現(xiàn)當(dāng)前線程有事務(wù)，則把Connection自動(dòng)提交設(shè)置為false,同時(shí)將該連接納入當(dāng)前事務(wù)中。
• 第三步：異趁氤啵回滾
  一旦拋出異常猪瞬，則需要進(jìn)行事務(wù)的回滾操作∪肜海回滾就是將當(dāng)前事務(wù)進(jìn)行回滾陈瘦，該事務(wù)的回滾會(huì)調(diào)用和它關(guān)聯(lián)的所有Connection的回滾。

3.1.2 Atomikos的分布式事務(wù)

代碼同上潮售，配置為：

    <bean id="atomikosUserTransaction" class="com.atomikos.icatch.jta.UserTransactionImp">
        <property name="transactionTimeout" value="300" />
    </bean>

    <bean id="springTransactionManager" class="org.springframework.transaction.jta.JtaTransactionManager">
        <property name="userTransaction" ref="atomikosUserTransaction" />
    </bean>

    <tx:annotation-driven transaction-manager="springTransactionManager"/>

    <!-- 配置數(shù)據(jù)源 -->
    <bean id="dataSourceC" class="com.atomikos.jdbc.AtomikosDataSourceBean" init-method="init" destroy-method="close">
        <property name="uniqueResourceName" value="XA1DBMS" />
        <property name="xaDataSourceClassName" value="com.mysql.jdbc.jdbc2.optional.MysqlXADataSource" />
        <property name="xaProperties">
            <props>
                <prop key="URL">jdbc:mysql://localhost:3306/test?useUnicode=true&amp;characterEncoding=utf-8</prop>
                <prop key="user">xxx</prop>
                <prop key="password">xxx</prop>
            </props>
        </property>
        <property name="poolSize" value="3" />
        <property name="minPoolSize" value="3" />
        <property name="maxPoolSize" value="5" />
    </bean>

    <!-- 配置數(shù)據(jù)源 -->
    <bean id="dataSourceD" class="com.atomikos.jdbc.AtomikosDataSourceBean" init-method="init" destroy-method="close">
        <property name="uniqueResourceName" value="XA2DBMS" />
        <property name="xaDataSourceClassName" value="com.mysql.jdbc.jdbc2.optional.MysqlXADataSource" />
        <property name="xaProperties">
            <props>
                <prop key="URL">jdbc:mysql://localhost:3306/test2?useUnicode=true&amp;characterEncoding=utf-8</prop>
                <prop key="user">xxx</prop>
                <prop key="password">xxx</prop>
            </props>
        </property>
        <property name="poolSize" value="3" />
        <property name="minPoolSize" value="3" />
        <property name="maxPoolSize" value="5" />
    </bean>

    <bean id="jdbcTemplateC" class="org.springframework.jdbc.core.JdbcTemplate">
        <property name="dataSource" ref="dataSourceC" />
    </bean>

    <bean id="jdbcTemplateD" class="org.springframework.jdbc.core.JdbcTemplate">
        <property name="dataSource" ref="dataSourceD" />
    </bean>

事務(wù)配置：
我們知道分布式事務(wù)中需要一個(gè)事務(wù)管理器即接口javax.transaction.TransactionManager痊项、面向開(kāi)發(fā)人員的javax.transaction.UserTransaction。對(duì)于Atomikos來(lái)說(shuō)分別對(duì)應(yīng)如下：

• com.atomikos.icatch.jta.UserTransactionImp
• com.atomikos.icatch.jta.UserTransactionManager
  我們?nèi)绻胧褂梅植际绞聞?wù)的同時(shí)酥诽，又想使用Spring帶給我們的@Transactional便利线婚，就需要配置一個(gè)JtaTransactionManager，而該JtaTransactionManager是需要一個(gè)userTransaction實(shí)例的

<bean id="userTransaction" class="com.atomikos.icatch.jta.UserTransactionImp">  
    <property name="transactionTimeout" value="300" />  
</bean>
<bean id="springTransactionManager" class="org.springframework.transaction.jta.JtaTransactionManager">  
    <property name="userTransaction" ref="userTransaction" />   
</bean>
<tx:annotation-driven transaction-manager="springTransactionManager"/>

可以對(duì)比下jotm的案例配置jotm的分布式事務(wù)配置盆均。可以看到j(luò)otm中使用的xapool中的StandardXADataSource是需要一個(gè)transactionManager的漱逸，而Atomikos使用的AtomikosNonXADataSourceBean則不需要泪姨。我們知道，StandardXADataSource中有了transactionManager就可以獲取當(dāng)前線程的事務(wù)饰抒，同時(shí)把XAResource加入進(jìn)當(dāng)前事務(wù)中去肮砾，而AtomikosNonXADataSourceBean卻沒(méi)有，它是怎么把XAResource加入進(jìn)當(dāng)前線程綁定的事務(wù)呢袋坑？這時(shí)候就需要可以通過(guò)靜態(tài)方法隨時(shí)獲取當(dāng)前線程綁定的事務(wù)仗处。
使用到的JAR包：

compile 'com.atomikos:transactions-jdbc:4.0.0M4'

3.2 單機(jī)事務(wù)+同步回調(diào)（異步）

以訂單子系統(tǒng)和支付子系統(tǒng)為例，如下圖：

Image text

如上圖枣宫，payment是支付系統(tǒng)婆誓，trade是訂單系統(tǒng)，兩個(gè)系統(tǒng)對(duì)應(yīng)的數(shù)據(jù)庫(kù)是分開(kāi)的也颤。支付完成之后洋幻，支付系統(tǒng)需要通知訂單系統(tǒng)狀態(tài)變更。
對(duì)于payment要執(zhí)行的操作可以用偽代碼表示如下：

begin tx;
  count = update account set amount = amount - ${cash} where uid = ${uid} and amount >= amount
  if (count <= 0) return false
    update payment_record set status = paid where trade_id = ${tradeId}
commit;

對(duì)于trade要執(zhí)行的操作可以用偽代碼表示如下：

begin tx;
  count = update trade_record set status = paid where trade_id = ${trade_id} and status = unpaid
  if (count <= 0) return false
    do other things ...
commit;

但是對(duì)于這兩段代碼如何串起來(lái)是個(gè)問(wèn)題翅娶，我們?cè)黾右粋€(gè)事務(wù)表文留，即圖中的tx_info，來(lái)記錄成功完成的支付事務(wù)竭沫，tx_info中需要有可以標(biāo)示被支付系統(tǒng)處理狀態(tài)的字段燥翅，為了和支付信息一致，需要放入事務(wù)中蜕提，代碼如下：

begin tx;
  count = update account set amount = amount - ${cash} where uid = ${uid} and amount >= amount
  if (count <= 0) return false
    update payment_record set status = paid where trade_id = ${tradeId}
    insert into tx_info values(${trade_id},${amount}...)
commit;

支付系統(tǒng)邊界到此為止森书，接下來(lái)就是訂單系統(tǒng)輪詢?cè)L問(wèn)tx_info，拉取已經(jīng)支付成功的訂單信息，對(duì)每一條信息都執(zhí)行trade系統(tǒng)的邏輯拄氯，偽代碼如下：

foreach trade_id in tx_info
  do trade_tx
  save tx_info.id to some store

事無(wú)延遲取決于時(shí)間程序輪詢間隔躲查，這樣我們做到了一致性，最終訂單都會(huì)在支付之后的最大時(shí)間間隔內(nèi)完成狀態(tài)遷移译柏。
當(dāng)然镣煮，這里也可以采用支付系統(tǒng)通過(guò)RPC方式同步通知訂單系統(tǒng)的方式來(lái)實(shí)現(xiàn)，處理狀態(tài)通過(guò)tx_info中的字段來(lái)表示鄙麦。
另外典唇，交易系統(tǒng)每次拉取數(shù)據(jù)的起點(diǎn)以及消費(fèi)記錄需要記錄下來(lái)，這樣才能不遺漏不重復(fù)地執(zhí)行胯府，所以需要增加一張表用于排重介衔，即上圖中的tx_duplication。但是每次對(duì)tx_duplication表的插入要在trade_tx的事務(wù)中完成骂因，偽代碼如下：

begin tx;
  c = insert ignore tx_duplication values($trade_id...)
  if (c <= 0) return false
    count = update trade_record set status = paid where trade_id = ${trade_id} and status = unpaid
  if (count <= 0) return false
    do other things ...
commit;

另外炎咖，tx_duplication表中trade_id表上必須有唯一鍵，這個(gè)算是結(jié)合之前的冪等篇來(lái)保證trade_tx的操作是冪等的寒波。

3.3 MQ做中間表角色

在上面的方案中乘盼，tx_info表所起到的作用就是隊(duì)列作用，記錄一個(gè)系統(tǒng)的表更俄烁，作為通知給需要感知的系統(tǒng)的事件绸栅。而時(shí)間程序去拉取只是系統(tǒng)去獲取感興趣事件的一個(gè)方式，而對(duì)應(yīng)交易系統(tǒng)的本地事務(wù)只是對(duì)應(yīng)消費(fèi)事件的一個(gè)過(guò)程页屠。在這樣的描述下粹胯，這些功能就是一個(gè)MQ——消息中間件。如下圖

Image text

  sendPrepare();
  isCommit = local_tx()
  if (isCommit) sendCommit()
    else sendRollback()

在做本地事務(wù)之前，先向MQ發(fā)送一個(gè)prepare消息由驹，然后執(zhí)行本地事務(wù)芍锚，本地事務(wù)提交成功的話昔园，向MQ發(fā)送一個(gè)commit消息，否則發(fā)送一個(gè)abort消息并炮，取消之前的消息默刚。MQ只會(huì)在收到commit確認(rèn)才會(huì)將消息投遞出去，所以這樣的形式可以保證在一切正常的情況下逃魄，本地事務(wù)和MQ可以達(dá)到一致性荤西。
但是分布式存在異常情況，網(wǎng)絡(luò)超時(shí)伍俘，機(jī)器宕機(jī)等等邪锌，比如當(dāng)系統(tǒng)執(zhí)行了local_tx()成功之后，還沒(méi)來(lái)得及將commit消息發(fā)送給MQ癌瘾，或者說(shuō)發(fā)送出去了觅丰，網(wǎng)絡(luò)超時(shí)了等等原因，MQ沒(méi)有收到commit妨退，即commit消息丟失了妇萄，那么MQ就不會(huì)把prepare消息投遞出去。如果這個(gè)無(wú)法保證的話咬荷，那么這個(gè)方案是不可行的嚣伐。針對(duì)這種情況，需要一個(gè)第三方異常校驗(yàn)?zāi)K來(lái)對(duì)MQ中在一定時(shí)間段內(nèi)沒(méi)有commit/abort 的消息和發(fā)消息的系統(tǒng)進(jìn)行檢查萍丐，確認(rèn)該消息是否應(yīng)該投遞出去或者丟棄，得到系統(tǒng)的確認(rèn)之后放典，MQ會(huì)做投遞還是丟棄逝变，這樣就完全保證了MQ和發(fā)消息的系統(tǒng)的一致性，從而保證了接收消息系統(tǒng)的一致性奋构。
這個(gè)方案要求MQ的系統(tǒng)可用性必須非常高壳影，至少要超過(guò)使用MQ的系統(tǒng)（推薦rocketmq，kafka都支持發(fā)送預(yù)備消息和業(yè)務(wù)回查）弥臼，這樣才能保證依賴他的系統(tǒng)能穩(wěn)定運(yùn)行宴咧。

3.4 SAGA方案

項(xiàng)目地址：https://github.com/apache/servicecomb-saga
Saga處理場(chǎng)景是要求相關(guān)的子事務(wù)提供事務(wù)處理函數(shù)同時(shí)也提供補(bǔ)償函數(shù)。Saga協(xié)調(diào)器alpha會(huì)根據(jù)事務(wù)的執(zhí)行情況向omega發(fā)送相關(guān)的指令径缅，確定是否向前重試或者向后恢復(fù)掺栅。

成功場(chǎng)景

成功場(chǎng)景下，每個(gè)事務(wù)都會(huì)有開(kāi)始和有對(duì)應(yīng)的結(jié)束事件纳猪。

Image text

異常場(chǎng)景

異常場(chǎng)景下氧卧，omega會(huì)向alpha上報(bào)中斷事件，然后alpha會(huì)向該全局事務(wù)的其它已完成的子事務(wù)發(fā)送補(bǔ)償指令氏堤，確保最終所有的子事務(wù)要么都成功沙绝，要么都回滾。

Image text

超時(shí)場(chǎng)景

超時(shí)場(chǎng)景下，已超時(shí)的事件會(huì)被alpha的定期掃描器檢測(cè)出來(lái)闪檬，與此同時(shí)星著，該超時(shí)事務(wù)對(duì)應(yīng)的全局事務(wù)也會(huì)被中斷。

Image text

例子

假設(shè)要租車(chē)粗悯、預(yù)訂酒店滿足分布式事務(wù)虚循。
租車(chē)服務(wù)

@Service
class CarBookingService {
  private Map<Integer, CarBooking> bookings = new ConcurrentHashMap<>();

  @Compensable(compensationMethod = "cancel")
  void order(CarBooking booking) {
    booking.confirm();
    bookings.put(booking.getId(), booking);
  }

  void cancel(CarBooking booking) {
    Integer id = booking.getId();
    if (bookings.containsKey(id)) {
      bookings.get(id).cancel();
    }
  }

  Collection<CarBooking> getAllBookings() {
    return bookings.values();
  }

  void clearAllBookings() {
    bookings.clear();
  }
}

酒店預(yù)訂

@Service
class HotelBookingService {
  private Map<Integer, HotelBooking> bookings = new ConcurrentHashMap<>();

  @Compensable(compensationMethod = "cancel")
  void order(HotelBooking booking) {
    if (booking.getAmount() > 2) {
      throw new IllegalArgumentException("can not order the rooms large than two");
    }
    booking.confirm();
    bookings.put(booking.getId(), booking);
  }

  void cancel(HotelBooking booking) {
    Integer id = booking.getId();
    if (bookings.containsKey(id)) {
      bookings.get(id).cancel();
    }
  }

  Collection<HotelBooking> getAllBookings() {
    return bookings.values();
  }

  void clearAllBookings() {
    bookings.clear();
  }
}

主服務(wù)

@RestController
public class BookingController {

  @Value("${car.service.address:http://car.servicecomb.io:8080}")
  private String carServiceUrl;

  @Value("${hotel.service.address:http://hotel.servicecomb.io:8080}")
  private String hotelServiceUrl;

  @Autowired
  private RestTemplate template;

  @SagaStart
  @PostMapping("/booking/{name}/{rooms}/{cars}")
  public String order(@PathVariable String name,  @PathVariable Integer rooms, @PathVariable Integer cars) {
    template.postForEntity(
        carServiceUrl + "/order/{name}/{cars}",
        null, String.class, name, cars);

    postCarBooking();

    template.postForEntity(
        hotelServiceUrl + "/order/{name}/{rooms}",
        null, String.class, name, rooms);

    postBooking();

    return name + " booking " + rooms + " rooms and " + cars + " cars OK";
  }

  // This method is used by the byteman to inject exception here
  private void postCarBooking() {

  }

  // This method is used by the byteman to inject the faults such as the timeout or the crash
  private void postBooking() {

  }
}

執(zhí)行流程

• 在Alpha目錄執(zhí)行 mvn clean package -DskipTests -Pdemo
• 執(zhí)行 java -Dspring.profiles.active=prd -D"spring.datasource.url=jdbc:postgresql://{saga_version}-exec.jar
• 在saga spring demo目錄執(zhí)行 mvn clean package -DskipTests -Pdemo
• java -Dserver.port=8081 -Dalpha.cluster.address={saga_version}-exec.jar
• java -Dserver.port=8082 -Dalpha.cluster.address={saga_version}-exec.jar
• java -Dserver.port=8083 -Dalpha.cluster.address={host_address}:8082 -Dhotel.service.address={saga_version}-exec.jar[alpha_address不帶http其他地址要帶上http]

3.5 TCC方案

項(xiàng)目地址https://github.com/QNJR-GROUP/EasyTransaction[對(duì)比tcc-transaction，Hmily为黎，ByteTCC來(lái)說(shuō)EasyTransaction性能最好邮丰，壓測(cè)未發(fā)現(xiàn)錯(cuò)誤], 當(dāng)然你也可以使用上面提到的SAGA項(xiàng)目，也是支持TCC協(xié)議的铭乾。下面我們舉個(gè)例子來(lái)看TCC是如何處理業(yè)務(wù)邏輯的剪廉。

eg：訂單支付

• 1：訂單服務(wù)->修改訂單狀態(tài)
• 2：庫(kù)存服務(wù)->扣減庫(kù)存
• 3：積分服務(wù)->增加積分
• 4：倉(cāng)庫(kù)服務(wù)->創(chuàng)建出庫(kù)單

try階段

• 1：訂單服務(wù)->狀態(tài)變更為“UpDating”
• 2：庫(kù)存服務(wù)->可用庫(kù)存減少1，凍結(jié)庫(kù)存增加1
• 3：積分服務(wù)->積分不變炕檩，增加預(yù)備積分
• 4：倉(cāng)庫(kù)服務(wù)->創(chuàng)建出庫(kù)單斗蒋，狀態(tài)設(shè)置為“UnKnown”

confirm階段

• 1：訂單服務(wù)->狀態(tài)變更為“已支付”
• 2：庫(kù)存服務(wù)->凍結(jié)庫(kù)存清零
• 3：積分服務(wù)->積分增加，預(yù)備積分清零
• 4：倉(cāng)庫(kù)服務(wù)->狀態(tài)設(shè)置為“出庫(kù)單已創(chuàng)建”

cancel階段

• 1：訂單服務(wù)->狀態(tài)變更為“已取消”
• 2：庫(kù)存服務(wù)->可用庫(kù)存增加笛质，凍結(jié)庫(kù)存清零
• 3：積分服務(wù)->預(yù)備積分清零
• 4：倉(cāng)庫(kù)服務(wù)->狀態(tài)設(shè)置為“已取消”

4 小結(jié)

業(yè)務(wù)各有各的不同偷霉，有些業(yè)務(wù)能容忍短期不一致爬虱，有些業(yè)務(wù)的操作可以冪等，無(wú)論什么樣的分布式事務(wù)解決方案都有其優(yōu)缺點(diǎn)腾它，沒(méi)有一個(gè)銀彈能夠適配所有。因此死讹，業(yè)務(wù)需要什么樣的解決方案瞒滴，還需要結(jié)合自身的業(yè)務(wù)需求、業(yè)務(wù)特點(diǎn)赞警、技術(shù)架構(gòu)以及各解決方案的特性妓忍，綜合分析，才能找到最適合的方案愧旦。