分布式事務(wù)解決方案

下面列舉一些分布式事務(wù)的解決方案：

• 2PC模式
• TCC事務(wù)補(bǔ)償
• 最大努力通知
• 可靠消息

2PC模式

2PC意為二階段提交，又叫XA Transactions，其中，XA是一個(gè)兩階段提交協(xié)議，該協(xié)議將事務(wù)分為兩個(gè)階段：

1. 第一階段：事務(wù)協(xié)調(diào)器要求每個(gè)涉及到事務(wù)的數(shù)據(jù)庫預(yù)提交此操作，并響應(yīng)是否可以提交
2. 第二階段：事務(wù)協(xié)調(diào)器要求每個(gè)數(shù)據(jù)庫提交數(shù)據(jù)，其中，如果有任何一個(gè)數(shù)據(jù)庫否決此次提交，那么所有的數(shù)據(jù)庫都會(huì)被要求回滾它們在此事務(wù)中修改的內(nèi)容

TCC事務(wù)補(bǔ)償

2PC模式遵循的是ACID原則，即：原子性、一致性、隔離性、持久性，它是一種強(qiáng)一致性的設(shè)計(jì)，而事實(shí)上，很多情況下我們都無法做到強(qiáng)一致性，或者想要實(shí)現(xiàn)強(qiáng)一致性成本比較高。eBay 的架構(gòu)師 Dan Pritchett在ACM上發(fā)表文章提出了BASE理論，該理論指出即使無法做到強(qiáng)一致性，但應(yīng)用可以采用合適的方式達(dá)到最終一致性，最終一致性對(duì)于數(shù)據(jù)的要求并不是非常嚴(yán)格，它不需要系統(tǒng)做到數(shù)據(jù)實(shí)時(shí)保持最新狀態(tài)，而是數(shù)據(jù)在經(jīng)過一段時(shí)間后，最終能夠達(dá)到一致即可，TCC事務(wù)補(bǔ)償方案就是一種柔性事務(wù)的設(shè)計(jì)，它能夠保證數(shù)據(jù)的最終一致性，一般是在業(yè)務(wù)層進(jìn)行實(shí)現(xiàn)的。如圖所示，我們必須在需要被事務(wù)控制的服務(wù)中編寫Try、Confirm、Cancel三個(gè)方法，此時(shí)，業(yè)務(wù)服務(wù)會(huì)嘗試調(diào)用每個(gè)服務(wù)的Try方法，若是沒有出現(xiàn)問題，則由事務(wù)管理器調(diào)用Confirm方法進(jìn)行提交，若是出現(xiàn)問題，則調(diào)用Cancel方法進(jìn)行回滾。TCC事務(wù)補(bǔ)償方案也比較好理解，但它對(duì)業(yè)務(wù)的侵入較大，且與業(yè)務(wù)耦合在了一起。

最大努力通知

最大努力通知方案也是柔性事務(wù)的設(shè)計(jì)，它按規(guī)律進(jìn)行消息通知，不保證數(shù)據(jù)一定能通知成功，但會(huì)提供可查詢接口進(jìn)行核對(duì)。如果你在項(xiàng)目中對(duì)接過支付寶支付服務(wù)，就應(yīng)該清楚，支付寶在付款后采用的就是最大努力通知方案，支付寶會(huì)每隔一段時(shí)間發(fā)送一個(gè)通知來告訴開發(fā)者訂單的支付情況，只有返回了 success?數(shù)據(jù)后支付寶才會(huì)停止通知。

可靠消息

可靠消息仍然只保證數(shù)據(jù)的最終一致性，且它需要借助消息中間件來完成，當(dāng)某個(gè)服務(wù)在事務(wù)提交之前，會(huì)向消息中間件發(fā)送一條消息，再根據(jù)本地事務(wù)的執(zhí)行狀態(tài)發(fā)送Commit或者RollBack給消息中間件，消費(fèi)方根據(jù)對(duì)應(yīng)的狀態(tài)對(duì)事務(wù)進(jìn)行對(duì)應(yīng)的處理。

RabbitMQ實(shí)現(xiàn)數(shù)據(jù)的最終一致性

想象一個(gè)場景，用戶在進(jìn)行下單操作之后，會(huì)有30分鐘的時(shí)間讓用戶進(jìn)行付款，在用戶付款之前，商品的庫存并沒有真正的扣除，而是進(jìn)行鎖定。若是用戶在規(guī)定時(shí)間內(nèi)付款成功，則需要真正扣減庫存；若是用戶在規(guī)定時(shí)間內(nèi)沒有付款，則需要將鎖定的商品庫存重新解鎖，這里涉及到兩個(gè)模塊之間的事務(wù)操作：當(dāng)用戶下單后，訂單模塊需要保存本次訂單的庫存工作單，記錄的是哪些商品需要被鎖定的庫存數(shù)，并遠(yuǎn)程調(diào)用庫存模塊鎖定庫存，此時(shí)需要等待用戶進(jìn)行付款，若是用戶未付款，則解鎖庫存。那么如何知道哪些訂單是用戶超時(shí)未付款需要解鎖的呢？我們可以編寫一個(gè)定時(shí)任務(wù)，讓其每隔一定的時(shí)間就去掃描訂單，并判斷是否超時(shí)，若是超時(shí)則解鎖庫存。不過這種方式并不理想，不僅效率低，而且掃描時(shí)間也不好確定。我們可以使用消息中間件來解決這一問題：訂單模塊在生成訂單后向RabbitMQ發(fā)送一條消息，并由RabbitMQ暫時(shí)保存，當(dāng)30分鐘過后，訂單過期，RabbitMQ再將消息提供給庫存模塊進(jìn)行消費(fèi)， 當(dāng)庫存模塊得到消息后就明白有訂單過期了，再去解鎖對(duì)應(yīng)的庫存即可，所以接下來我們面臨的問題就是：如何將消息在RabbitMQ中保存30分鐘，當(dāng)消息過期后再交給庫存模塊消費(fèi)。

延時(shí)隊(duì)列

在RabbitMQ中，我們可以實(shí)現(xiàn)一個(gè)延時(shí)隊(duì)列，消息進(jìn)入延時(shí)隊(duì)列后不會(huì)立馬被消費(fèi)，而是需要等待設(shè)定的時(shí)間，在實(shí)現(xiàn)之前，需要清楚兩個(gè)概念：

• 死信
• 死信路由

死信

在RabbitMQ中，我們可以為消息設(shè)置一個(gè)存活時(shí)間TTL（time to live），當(dāng)消息超過了存活時(shí)間，就可以認(rèn)為這個(gè)消息已經(jīng)死了，稱為 死信?。一個(gè)消息在滿足如下條件時(shí)會(huì)進(jìn)入死信路由：

1. 消息被Consumer拒收，并且reject方法的參數(shù)中requeue值為false，即：該消息被拒收后，不會(huì)再進(jìn)入消息隊(duì)列
2. 消息超過了TTL時(shí)間，導(dǎo)致消息過期
3. 消息隊(duì)列滿了，排在前面的消息會(huì)被丟棄或者扔到死信路由上

死信路由

在死信概念中，我們一直強(qiáng)調(diào)一個(gè)詞， 死信路由?，其實(shí)，它就是一個(gè)普通的路由，只是當(dāng)某個(gè)隊(duì)列綁定了死信路由后，該消息隊(duì)列中的消息過期了，就會(huì)自動(dòng)觸發(fā)消息的轉(zhuǎn)發(fā)，消息會(huì)被扔到死信路由上。通過死信和死信路由，我們就能夠?qū)崿F(xiàn)一個(gè)延時(shí)隊(duì)列，如圖所示：當(dāng)生產(chǎn)者生產(chǎn)了一個(gè)消息后，會(huì)通過交換器放入一個(gè)隊(duì)列，該隊(duì)列比較特殊，在該隊(duì)列中的消息存活時(shí)間均為30分鐘，并且當(dāng)這些消息過期成為死信后，會(huì)被交給死信路由，死信路由再將消息放入另一個(gè)消息隊(duì)列，這樣，該消息隊(duì)列便保證了每次放入的消息都經(jīng)過了30分鐘，因?yàn)橹挥兴佬挪拍苓M(jìn)入該隊(duì)列，要想實(shí)現(xiàn)這一過程，我們需要為那個(gè)特殊的消息隊(duì)列設(shè)置一些屬性值：

• x-dead-letter-exchange：xxx ???? ??設(shè)置死信路由
• x-dead-letter-routing-key：xxx ??? 設(shè)置死信路由鍵
• x-message-ttl：1800 000? ??? ??? ??? ?設(shè)置消息的存活時(shí)間為30分鐘

將其類比到具體的業(yè)務(wù)場景中，就比如訂單超時(shí)自動(dòng)解鎖庫存的需求，其設(shè)計(jì)如下圖所示：執(zhí)行流程如下：

1. Publisher發(fā)送消息給路由order.delay.exchange，路由鍵為order.delay
2. 路由order.delay.exchange將消息放入綁定關(guān)系為order.delay的消息隊(duì)列order.delay.queue
3. 消息隊(duì)列order.delay.queue中消息的存活時(shí)間為30分鐘，當(dāng)消息過期后，消息會(huì)交給路由order.exchange，路由鍵為order
4. 路由order.exchange將消息放入綁定關(guān)系為order的消息隊(duì)列order.queue
5. Consumer監(jiān)聽消息隊(duì)列order.queue，保證了消費(fèi)的消息均是過期的

不過這一過程還可以再簡化一下：它與剛才唯一的區(qū)別在于少了一個(gè)路由，Publisher在將消息發(fā)送給路由order.delay.exchange之后，會(huì)將消息放入隊(duì)列order.delay.queue，我們讓該隊(duì)列在消息過期后仍然將消息交給路由order.delay.exchange，這樣就節(jié)省了一個(gè)路由的資源。

代碼實(shí)現(xiàn)

概念吹得天花亂墜，不動(dòng)手實(shí)現(xiàn)一下始終是無法深刻理解的，所以，我們就通過一個(gè)案例來感受一下延遲隊(duì)列的效果。

創(chuàng)建一個(gè)SpringBoot應(yīng)用，并使用代碼創(chuàng)建出路由、消息隊(duì)列及其它們之間的關(guān)系：

@Configuration
public?class?MyRabbitMQConfig?{

????@Bean
????public?Queue?orderDelayQueue()?{
????????Map?arguments?=?new?HashMap<>();
????????arguments.put("x-dead-letter-exchange",?"order.delay.exchange");
????????arguments.put("x-dead-letter-routing-key",?"order.release.order");
????????arguments.put("x-message-ttl",?1000?*?10);
????????return?new?Queue("order.delay.queue",?true,?false,?false,?arguments);
????}

????@Bean
????public?Queue?orderReleaseOrderQueue()?{
????????return?new?Queue("order.release.order.queue",?true,?false,?false);
????}

????@Bean
????public?Exchange?orderDelayExchange()?{
????????return?new?TopicExchange("order.delay.exchange",?true,?false);
????}

????@Bean
????public?Binding?orderCreateOrderBinding()?{
????????return?new?Binding("order.delay.queue",
???????????????????????????Binding.DestinationType.QUEUE,
???????????????????????????"order.delay.exchange",
???????????????????????????"order.create.order",?null);
????}

????@Bean
????public?Binding?orderReleaseOrderBinding()?{
????????return?new?Binding("order.release.order.queue",
???????????????????????????Binding.DestinationType.QUEUE,
???????????????????????????"order.delay.exchange",
???????????????????????????"order.release.order",?null);
????}
}

需要注意的是使用這種方式創(chuàng)建需要發(fā)送一下消息：

@RestController
public?class?TestController?{

????@Autowired
????private?RabbitTemplate?rabbitTemplate;

????@GetMapping("/test")
????public?String?test(){
????????rabbitTemplate.convertAndSend("order.delay.queue","message");
????????return?"test";
????}
}

當(dāng)發(fā)送消息到隊(duì)列 order.delay.queue?時(shí)，RabbitMQ便會(huì)創(chuàng)建出隊(duì)列和路由：接下來我們編寫一個(gè)監(jiān)聽方法，它用來監(jiān)聽隊(duì)列 order.release.order.queue?：

@RabbitListener(queues?=?"order.release.order.queue")
public?void?listener(Message?message)?{
????System.out.println("收到消息："?+?message);
}

此時(shí)我們訪問 http://localhost:8080/test ，就會(huì)發(fā)送一條消息到隊(duì)列，再經(jīng)過10秒鐘的時(shí)間，消息會(huì)過期，消息便會(huì)進(jìn)入隊(duì)列 order.release.order.queue?，而我們監(jiān)聽的又是這個(gè)隊(duì)列，所以總能收到10秒后過期的消息：

收到消息：(Body:'message'?......)
收到消息：(Body:'message'?......)
收到消息：(Body:'message'?......)

通過這樣的方式，我們便能夠?qū)崿F(xiàn)數(shù)據(jù)的最終一致性。