分布式事務基礎
事務
事務指的就是一個操作單元，在這個操作單元中的所有操作最終要保持一致的行為，要么所有操作都成功，要么所有的操作都被撤銷。簡單地說，事務提供一種“要么什么都不做，要么做全套”機制。
本地事務
本地事物其實可以認為是數據庫提供的事務機制。說到數據庫事務就不得不說，數據庫事務中的四大特性:
- A：原子性(Atomicity)，一個事務中的所有操作，要么全部完成，要么全部不完成
- C：一致性(Consistency)，在一個事務執行之前和執行之后數據庫都必須處於一致性狀態
- I：隔離性(Isolation)，在並發環境中，當不同的事務同時操作相同的數據時，事務之間互不影響
- D：持久性(Durability)，指的是只要事務成功結束，它對數據庫所做的更新就必須永久地保存下來
數據庫事務在實現時會將一次事務涉及的所有操作全部納入到一個不可分割的執行單元，該執行單元中的所有操作要么都成功，要么都失敗，只要其中任一操作執行失敗，都將導致整個事務的回滾
分布式事務
分布式事務指事務的參與者、支持事務的服務器、資源服務器以及事務管理器分別位於不同的分布式系統的不同節點之上。
簡單地說，就是一次大的操作由不同的小操作組成，這些小的操作分布在不同的服務器上，且屬於不同的應用，分布式事務需要保證這些小操作要么全部成功，要么全部失敗。
本質上來說，分布式事務就是為了保證不同數據庫的數據一致性。
分布式事務的場景
單體系統訪問多個數據庫
一個服務需要調用多個數據庫實例完成數據的增刪改操作
多個微服務訪問同一個數據庫
多個服務需要調用一個數據庫實例完成數據的增刪改操作
多個微服務訪問多個數據庫
多個服務需要調用一個數據庫實例完成數據的增刪改操作
分布式事務解決方案
全局事務
全局事務基於DTP模型實現。DTP是由X/Open組織提出的一種分布式事務模型——X/Open Distributed Transaction Processing Reference Model。它規定了要實現分布式事務，需要三種角色：
- AP: Application 應用系統 (微服務)
- TM: Transaction Manager 事務管理器 (全局事務管理)
- RM: Resource Manager 資源管理器 (數據庫)
整個事務分成兩個階段:
- 階段一: 表決階段，所有參與者都將本事務執行預提交，並將能否成功的信息反饋發給協調者。
- 階段二: 執行階段，協調者根據所有參與者的反饋，通知所有參與者，步調一致地執行提交或者回滾。
優點
- 提高了數據一致性的概率，實現成本較低
缺點
- 單點問題: 事務協調者宕機
- 同步阻塞: 延遲了提交時間，加長了資源阻塞時間
- 數據不一致: 提交第二階段，依然存在commit結果未知的情況，有可能導致數據不一致
可靠消息服務
基於可靠消息服務的方案是通過消息中間件保證上、下游應用數據操作的一致性。假設有A和B兩個系統，分別可以處理任務A和任務B。此時存在一個業務流程，需要將任務A和任務B在同一個事務中處理。就可以使用消息中間件來實現這種分布式事務。
第一步：消息由系統A投遞到中間件
- 在系統A處理任務A前，首先向消息中間件發送一條消息
- 消息中間件收到后將該條消息持久化，但並不投遞。持久化成功后，向A回復一個確認應答
- 系統A收到確認應答后，則可以開始處理任務A
- 任務A處理完成后，向消息中間件發送Commit或者Rollback請求。該請求發送完成后，對系統A而言，該事務的處理過程就結束了
- 如果消息中間件收到Commit，則向B系統投遞消息；如果收到Rollback，則直接丟棄消息。但是如果消息中間件收不到Commit和Rollback指令，那么就要依靠"超時詢問機制"。
超時詢問機制
系統A除了實現正常的業務流程外，還需提供一個事務詢問的接口，供消息中間件調用。當消息中間件收到發布消息便開始計時，如果到了超時沒收到確認指令，就會主動調用系統A提供的事務詢問接口詢問該系統目前的狀態。該接口會返回三種結果，中間件根據三種結果做出不同反應：
提交:將該消息投遞給系統B
回滾:直接將條消息丟棄
處理中:繼續等待
第二步:消息由中間件投遞到系統B
消息中間件向下游系統投遞完消息后便進入阻塞等待狀態，下游系統便立即進行任務的處理，任務處理完成后便向消息中間件返回應答。
- 如果消息中間件收到確認應答后便認為該事務處理完畢
- 如果消息中間件在等待確認應答超時之后就會重新投遞，直到下游消費者返回消費成功響應為止。
一般消息中間件可以設置消息重試的次數和時間間隔，如果最終還是不能成功投遞，則需要手工干預。這里之所以使用人工干預，而不是使用讓Ａ系統回滾，主要是考慮到整個系統設計的復雜度問題。
基於可靠消息服務的分布式事務，前半部分使用異步，注重性能；后半部分使用同步，注重開發成本。
最大努力通知
最大努力通知也被稱為定期校對，其實是對第二種解決方案的進一步優化。它引入了本地消息表來記錄錯誤消息，然后加入失敗消息的定期校對功能，來進一步保證消息會被下游系統消費。
第一步:消息由系統A投遞到中間件
- 處理業務的同一事務中，向本地消息表中寫入一條記錄
- 准備專門的消息發送者不斷地發送本地消息表中的消息到消息中間件，如果發送失敗則重試
第二步:消息由中間件投遞到系統B
- 消息中間件收到消息后負責將該消息同步投遞給相應的下游系統，並觸發下游系統的任務執行
- 當下游系統處理成功后，向消息中間件反饋確認應答，消息中間件便可以將該條消息刪除，從而該事務完成
- 對於投遞失敗的消息，利用重試機制進行重試，對於重試失敗的，寫入錯誤消息表
- 消息中間件需要提供失敗消息的查詢接口，下游系統會定期查詢失敗消息，並將其消費
這種方式的優缺點：
- 優點： 一種非常經典的實現，實現了最終一致性。
- 缺點： 消息表會耦合到業務系統中，如果沒有封裝好的解決方案，會有很多雜活需要處理。
TCC事務
TCC即為Try Confifirm Cancel，它屬於補償型分布式事務。TCC實現分布式事務一共有三個步驟：
- Try： 嘗試待執行的業務：這個過程並未執行業務，只是完成所有業務的一致性檢查，並預留好執行所需的全部資源
- Confifirm： 確認執行業務：確認執行業務操作，不做任何業務檢查， 只使用Try階段預留的業務資源。通常情況下，采用TCC則認為 Confifirm階段是不會出錯的。即：只要Try成功，Confifirm一定成功。若Confifirm階段真的出錯了，需引入重試機制或人工處理。
- Cancel： 取消待執行的業務:取消Try階段預留的業務資源。通常情況下，采用TCC則認為Cancel階段也是一定成功的。若Cancel階段真的出錯了，需引入重試機制或人工處理
TCC兩階段提交與XA兩階段提交的區別是：
- XA是資源層面的分布式事務，強一致性，在兩階段提交的整個過程中，一直會持有資源的鎖。
- TCC是業務層面的分布式事務，最終一致性，不會一直持有資源的鎖。
TCC事務的優缺點：
- 優點：把數據庫層的二階段提交上提到了應用層來實現，規避了數據庫層的2PC性能低下問題。
- 缺點：TCC的Try、Confifirm和Cancel操作功能需業務提供，開發成本高。
Seata介紹
2019 年 1 月，阿里巴巴中間件團隊發起了開源項目 Fescar（Fast & EaSy Commit AndRollback），其願景是讓分布式事務的使用像本地事務的使用一樣，簡單和高效，並逐步解決開發者們遇到的分布式事務方面的所有難題。后來更名為 Seata，意為：Simple Extensible Autonomous Transaction Architecture，是一套分布式事務解決方案。
Seata的設計目標是對業務無侵入，因此從業務無侵入的2PC方案着手，在傳統2PC的基礎上演進。它把一個分布式事務理解成一個包含了若干分支事務的全局事務。全局事務的職責是協調其下管轄的分支事務達成一致，要么一起成功提交，要么一起失敗回滾。此外，通常分支事務本身就是一個關系數據庫的本地事務。
Seata主要由三個重要組件組成：
- TC：Transaction Coordinator 事務協調器，管理全局的分支事務的狀態，用於全局性事務的提交和回滾。
- TM：Transaction Manager 事務管理器，用於開啟、提交或者回滾全局事務。
- RM：Resource Manager 資源管理器，用於分支事務上的資源管理，向TC注冊分支事務，上報分支事務的狀態，接受TC的命令來提交或者回滾分支事務。
Seata的執行流程如下:
- A服務的TM向TC申請開啟一個全局事務，TC就會創建一個全局事務並返回一個唯一的XID
- A服務的RM向TC注冊分支事務，並及其納入XID對應全局事務的管轄
- A服務執行分支事務，向數據庫做操作
- A服務開始遠程調用B服務，此時XID會在微服務的調用鏈上傳播
- B服務的RM向TC注冊分支事務，並將其納入XID對應的全局事務的管轄
- B服務執行分支事務，向數據庫做操作
- 全局事務調用鏈處理完畢，TM根據有無異常向TC發起全局事務的提交或者回滾
- TC協調其管轄之下的所有分支事務， 決定是否回滾
Seata實現2PC與傳統2PC的差別：
- 架構層次方面，傳統2PC方案的 RM 實際上是在數據庫層，RM本質上就是數據庫自身，通過XA協議實現，而 Seata的RM是以jar包的形式作為中間件層部署在應用程序這一側的。
- 兩階段提交方面，傳統2PC無論第二階段的決議是commit還是rollback，事務性資源的鎖都要保持到Phase2完成才釋放。而Seata的做法是在Phase1 就將本地事務提交，這樣就可以省去Phase2持鎖的時間，整體提高效率
Seata實現分布式事務控制
本示例通過Seata中間件實現分布式事務，模擬電商中的下單和扣庫存的過程
我們通過訂單微服務執行下單操作，然后由訂單微服務調用商品微服務扣除庫存
案例基本代碼
修改order微服務
controller
@RestController @Slf4j public class OrderController5 { @Autowired private OrderServiceImpl5 orderService; //下單 @RequestMapping("/order/prod/{pid}") public Order order(@PathVariable("pid") Integer pid) { log.info("接收到{}號商品的下單請求,接下來調用商品微服務查詢此商品信息", pid); return orderService.createOrder(pid); } } OrderService
@Service @Slf4j public class OrderServiceImpl5{ @Autowired private OrderDao orderDao; @Autowired private ProductService productService; @Autowired private RocketMQTemplate rocketMQTemplate; @GlobalTransactional public Order createOrder(Integer pid) { //1 調用商品微服務,查詢商品信息 Product product = productService.findByPid(pid); log.info("查詢到{}號商品的信息,內容是:{}", pid, JSON.toJSONString(product)); //2 下單(創建訂單) Order order = new Order(); order.setUid(1); order.setUsername("測試用戶"); order.setPid(pid); order.setPname(product.getPname()); order.setPprice(product.getPprice()); order.setNumber(1); orderDao.save(order); log.info("創建訂單成功,訂單信息為{}", JSON.toJSONString(order)); //3 扣庫存 productService.reduceInventory(pid, order.getNumber()); //4 向mq中投遞一個下單成功的消息 rocketMQTemplate.convertAndSend("order-topic", order); return order; } } ProductService
@FeignClient(value = "service-product") public interface ProductService { //減庫存 @RequestMapping("/product/reduceInventory") void reduceInventory(@RequestParam("pid") Integer pid, @RequestParam("num") int num); } 修改Product微服務
controller
//減少庫存 @RequestMapping("/product/reduceInventory") public void reduceInventory(Integer pid, int num) { productService.reduceInventory(pid, num); } service
@Override public void reduceInventory(Integer pid, int num) { Product product = productDao.findById(pid).get(); product.setStock(product.getStock() - num); //減庫存 productDao.save(product); } 異常模擬
在ProductServiceImpl的代碼中模擬一個異常, 然后調用下單接口
@Override public void reduceInventory(Integer pid, Integer number) { Product product = productDao.findById(pid).get(); if (product.getStock() < number) { throw new RuntimeException("庫存不足"); } int i = 1 / 0; product.setStock(product.getStock() - number); productDao.save(product); } 啟動Seata
修改配置文件
將下載得到的壓縮包進行解壓，進入conf目錄，調整下面的配置文件：
registry.conf
registry { type = "nacos" nacos { serverAddr = "localhost" namespace = "public" cluster = "default" } } config { type = "nacos" nacos { serverAddr = "localhost" namespace = "public" cluster = "default" } } nacos-confifig.txt
service.vgroup_mapping.service-product=default service.vgroup_mapping.service-order=default 這里的語法為： service.vgroup_mapping.${your-service-gruop}=default，中間的${your-service-gruop}為自己定義的服務組名稱， 這里需要我們在程序的配置文件中配置
初始化seata在nacos的配置
\# 初始化seata 的nacos配置 \# 注意: 這里要保證nacos是已經正常運行的 cd conf nacos-config.sh 127.0.0.1 執行成功后可以打開Nacos的控制台，在配置列表中，可以看到初始化了很多Group為SEATA_GROUP的配置。
啟動seata服務
cd bin seata-server.bat -p 9000 -m file 啟動后在 Nacos 的服務列表下面可以看到一個名為 serverAddr 的服務。
使用Seata實現事務控制
初始化數據表
在我們的數據庫中加入一張undo_log表,這是Seata記錄事務日志要用到的表
CREATE TABLE `undo_log` ( `id` BIGiNT(20) NOT NULL AUTO_INCREMENT, `branch_id` BIGiNT(20) NOT NULL, `xid` VARcHAR(100) NOT NULL, `context` VARcHAR(128) NOT NULL, `rollback_info` LONGBLOB NOT NULL, `log_status` iNT(11) NOT NULL, `log_created` DATETIME NOT NULL, `log_modified` DATETIME NOT NULL, `ext` VARcHAR(100) DEFAULT NULL, PRIMARY KEY (`id`), UNIQUE KEY `ux_undo_log` (`xid`, `branch_id`) ) ENGINE = INNODB AUTO_INCREMENT = 1 DEFAULT CHARSET = utf8; 添加配置
在需要進行分布式控制的微服務中進行下面幾項配置:
添加依賴
<dependency> <groupId>com.alibaba.cloud</groupId> <artifactId>spring-cloud-starter-alibaba-seata</artifactId> </dependency> <dependency> <groupId>com.alibaba.cloud</groupId> <artifactId>spring-cloud-starter-alibaba-nacos-config</artifactId> </dependency> DataSourceProxyConfifig
Seata 是通過代理數據源實現事務分支的，所以需要配置 io.seata.rm.datasource.DataSourceProxy 的Bean，且是 @Primary默認的數據源，否則事務不會回滾，無法實現分布式事務
@Configuration public class DataSourceProxyConfig { @Bean @ConfigurationProperties(prefix = "spring.datasource") public DruidDataSource druidDataSource() { return new DruidDataSource(); } @Primary @Bean public DataSourceProxy dataSource(DruidDataSource druidDataSource) { return new DataSourceProxy(druidDataSource); } } registry.conf
在resources下添加Seata的配置文件 registry.conf
registry { type = "nacos" nacos { serverAddr = "localhost" namespace = "public" cluster = "default" } } config { type = "nacos" nacos { serverAddr = "localhost" namespace = "public" cluster = "default" } } bootstrap.yaml
spring: application: name: service-product cloud: nacos: config: server-addr: localhost:8848 # nacos的服務端地址 namespace: public group: SEATA_GROUP alibaba: seata: tx-service-group: ${ spring.application.name } 在order微服務開啟全局事務
@GlobalTransactional//全局事務控制 public Order createOrder(Integer pid) {} 測試
再次下單測試
seata運行流程分析
要點說明：
- 每個RM使用DataSourceProxy連接數據庫，其目的是使用ConnectionProxy，使用數據源和數據連接代理的目的就是在第一階段將undo_log和業務數據放在一個本地事務提交，這樣就保存了只要有業務操作就一定有undo_log。
- 在第一階段undo_log中存放了數據修改前和修改后的值，為事務回滾作好准備，所以第一階段完成就已經將分支事務提交，也就釋放了鎖資源。
- TM開啟全局事務開始，將XID全局事務id放在事務上下文中，通過feign調用也將XID傳入下游分支事務，每個分支事務將自己的Branch ID分支事務ID與XID關聯。
- 第二階段全局事務提交，TC會通知各各分支參與者提交分支事務，在第一階段就已經提交了分支事務，這里各各參與者只需要刪除undo_log即可，並且可以異步執行，第二階段很快可以完成。
- 第二階段全局事務回滾，TC會通知各各分支參與者回滾分支事務，通過 XID 和 Branch ID 找到相應的回滾日志，通過回滾日志生成反向的 SQL 並執行，以完成分支事務回滾到之前的狀態，如果回滾失敗則會重試回滾操作