點(diǎn)擊上方藍(lán)色字體，選擇“標(biāo)星公眾號(hào)”
優(yōu)質(zhì)文章，第一時(shí)間送達(dá)
1、事務(wù)消費(fèi)介紹
我們經(jīng)常支付寶轉(zhuǎn)賬余額寶，這是日常生活的一件普通小事，但是我們思考支付寶扣除轉(zhuǎn)賬的錢之后，如果系統(tǒng)掛掉怎么辦，這時(shí)余額寶賬戶并沒有增加相應(yīng)的金額，數(shù)據(jù)就會(huì)出現(xiàn)不一致狀況了。
上述場(chǎng)景在各個(gè)類型的系統(tǒng)中都能找到相似影子，比如在電商系統(tǒng)中，當(dāng)有用戶下單后，除了在訂單表插入一條記錄外，對(duì)應(yīng)商品表的這個(gè)商品數(shù)量必須減1吧，怎么保證？！在搜索廣告系統(tǒng)中，當(dāng)用戶點(diǎn)擊某廣告后，除了在點(diǎn)擊事件表中增加一條記錄外，還得去商家賬戶表中找到這個(gè)商家并扣除廣告費(fèi)吧，怎么保證？！等等，相信大家或多或多少都能碰到相似情景。
本質(zhì)上問題可以抽象為：當(dāng)一個(gè)表數(shù)據(jù)更新后，怎么保證另一個(gè)表的數(shù)據(jù)也必須要更新成功。
如果是單機(jī)系統(tǒng)(數(shù)據(jù)庫實(shí)例也在同一個(gè)系統(tǒng)上)的話，我們可以用本地事務(wù)輕松解決：
還是以支付寶轉(zhuǎn)賬余額寶為例（比如轉(zhuǎn)賬10000塊錢），假設(shè)有
支付寶賬戶表：A（id，userId，amount）
余額寶賬戶表：B（id，userId，amount）
用戶的userId=1；
從支付寶轉(zhuǎn)賬1萬塊錢到余額寶的動(dòng)作分為兩步：
1）支付寶表扣除1萬：update A set amount=amount-10000 where userId=1;
2）余額寶表增加1萬：update B set amount=amount+10000 where userId=1;
如何確保支付寶余額寶收支平衡呢？
有人說這個(gè)很簡(jiǎn)單嘛，可以用事務(wù)解決。
Begin transaction
update A set amount=amount-10000 where userId=1;
update B set amount=amount+10000 where userId=1;
End transaction
commit;
這樣確實(shí)能解決，如果你使用spring的話一個(gè)注解就能搞定上述事務(wù)功能。
@Transactional(rollbackFor=Exception.class)
public void update() {
//更新A表
updateATable();
//更新B表
updateBTable();
}
如果系統(tǒng)規(guī)模較小，數(shù)據(jù)表都在一個(gè)數(shù)據(jù)庫實(shí)例上，上述本地事務(wù)方式可以很好地運(yùn)行，但是如果系統(tǒng)規(guī)模較大，比如支付寶賬戶表和余額寶賬戶表顯然不會(huì)在同一個(gè)數(shù)據(jù)庫實(shí)例上，他們往往分布在不同的物理節(jié)點(diǎn)上，這時(shí)本地事務(wù)已經(jīng)失去用武之地。
下面我們來看看比較主流的兩種方案：
2、分布式事務(wù)—————— 兩階段提交協(xié)議
兩階段提交協(xié)議（Two-phase Commit，2PC）經(jīng)常被用來實(shí)現(xiàn)分布式事務(wù)。一般分為協(xié)調(diào)器TC和若干事務(wù)執(zhí)行者兩種角色，這里的事務(wù)執(zhí)行者就是具體的數(shù)據(jù)庫，協(xié)調(diào)器可以和事務(wù)執(zhí)行器在一臺(tái)機(jī)器上。
我們根據(jù)上面的圖來看看主要流程：
1） 我們的應(yīng)用程序（client）發(fā)起一個(gè)開始請(qǐng)求到TC（transaction）；
2） TC先將prepare消息寫到本地日志，之后向所有的Si發(fā)起prepare消息。以支付寶轉(zhuǎn)賬到余額寶為例，TC給A的prepare消息是通知支付寶數(shù)據(jù)庫相應(yīng)賬目扣款1萬，TC給B的prepare消息是通知余額寶數(shù)據(jù)庫相應(yīng)賬目增加1w。為什么在執(zhí)行任務(wù)前需要先寫本地日志，主要是為了故障后恢復(fù)用，本地日志起到現(xiàn)實(shí)生活中憑證的效果，如果沒有本地日志（憑證），出問題容易死無對(duì)證；
3） Si收到prepare消息后，執(zhí)行具體本機(jī)事務(wù)，但不會(huì)進(jìn)行commit，如果成功返回yes，不成功返回no。同理，返回前都應(yīng)把要返回的消息寫到日志里，當(dāng)作憑證。
4） TC收集所有執(zhí)行器返回的消息，如果所有執(zhí)行器都返回yes，那么給所有執(zhí)行器發(fā)生送commit消息，執(zhí)行器收到commit后執(zhí)行本地事務(wù)的commit操作；如果有任一個(gè)執(zhí)行器返回no，那么給所有執(zhí)行器發(fā)送abort消息，執(zhí)行器收到abort消息后執(zhí)行事務(wù)abort操作。
注：TC或Si把發(fā)送或接收到的消息先寫到日志里，主要是為了故障后恢復(fù)用。如某一Si從故障中恢復(fù)后，先檢查本機(jī)的日志，如果已收到commit，則提交，如果abort則回滾。如果是yes，則再向TC詢問一下，確定下一步。如果什么都沒有，則很可能在prepare階段Si就崩潰了，因此需要回滾。
現(xiàn)如今實(shí)現(xiàn)基于兩階段提交的分布式事務(wù)也沒那么困難了，如果使用java，那么可以使用開源軟件atomikos(http://www.atomikos.com/)，來快速實(shí)現(xiàn)。)
不過但凡使用過的上述兩階段提交的同學(xué)都可以發(fā)現(xiàn)性能實(shí)在是太差，根本不適合高并發(fā)的系統(tǒng)。為什么？
1）兩階段提交涉及多次節(jié)點(diǎn)間的網(wǎng)絡(luò)通信，通信時(shí)間太長！
2）事務(wù)時(shí)間相對(duì)于變長了，鎖定的資源的時(shí)間也變長了，造成資源等待時(shí)間也增加好多！
正是由于分布式事務(wù)存在很嚴(yán)重的性能問題，大部分高并發(fā)服務(wù)都在避免使用，往往通過其他途徑來解決數(shù)據(jù)一致性問題。
3、使用消息隊(duì)列來避免分布式事務(wù)
如果仔細(xì)觀察生活的話，生活的很多場(chǎng)景已經(jīng)給了我們提示。
比如在北京很有名的姚記炒肝點(diǎn)了炒肝并付了錢后，他們并不會(huì)直接把你點(diǎn)的炒肝給你，而是給你一張小票，然后讓你拿著小票到出貨區(qū)排隊(duì)去取。為什么他們要將付錢和取貨兩個(gè)動(dòng)作分開呢？原因很多，其中一個(gè)很重要的原因是為了使他們接待能力增強(qiáng)（并發(fā)量更高）。
還是回到我們的問題，只要這張小票在，你最終是能拿到炒肝的。同理轉(zhuǎn)賬服務(wù)也是如此，當(dāng)支付寶賬戶扣除1萬后，我們只要生成一個(gè)憑證（消息）即可，這個(gè)憑證（消息）上寫著“讓余額寶賬戶增加1萬”，只要這個(gè)憑證（消息）能可靠保存，我們最終是可以拿著這個(gè)憑證（消息）讓余額寶賬戶增加1萬的，即我們能依靠這個(gè)憑證（消息）完成最終一致性。
那么我們?nèi)绾慰煽勘４鎽{證（消息）有兩種方法：
1）業(yè)務(wù)與消息耦合的方式
支付寶在完成扣款的同時(shí)，同時(shí)記錄消息數(shù)據(jù)，這個(gè)消息數(shù)據(jù)與業(yè)務(wù)數(shù)據(jù)保存在同一數(shù)據(jù)庫實(shí)例里（消息記錄表表名為message）。
Begin transaction
update A set amount=amount-10000 where userId=1;
insert into message(userId, amount,status) values(1, 10000, 1);
End transaction
commit;
上述事務(wù)能保證只要支付寶賬戶里被扣了錢，消息一定能保存下來。
當(dāng)上述事務(wù)提交成功后，我們通過實(shí)時(shí)消息服務(wù)將此消息通知余額寶，余額寶處理成功后發(fā)送回復(fù)成功消息，支付寶收到回復(fù)后刪除該條消息數(shù)據(jù)。
2）業(yè)務(wù)與消息解耦方式
上述保存消息的方式使得消息數(shù)據(jù)和業(yè)務(wù)數(shù)據(jù)緊耦合在一起，從架構(gòu)上看不夠優(yōu)雅，而且容易誘發(fā)其他問題。為了解耦，可以采用以下方式。
a）支付寶在扣款事務(wù)提交之前，向?qū)崟r(shí)消息服務(wù)請(qǐng)求發(fā)送消息，實(shí)時(shí)消息服務(wù)只記錄消息數(shù)據(jù)，而不真正發(fā)送，只有消息發(fā)送成功后才會(huì)提交事務(wù)；
b）當(dāng)支付寶扣款事務(wù)被提交成功后，向?qū)崟r(shí)消息服務(wù)確認(rèn)發(fā)送。只有在得到確認(rèn)發(fā)送指令后，實(shí)時(shí)消息服務(wù)才真正發(fā)送該消息；
c）當(dāng)支付寶扣款事務(wù)提交失敗回滾后，向?qū)崟r(shí)消息服務(wù)取消發(fā)送。在得到取消發(fā)送指令后，該消息將不會(huì)被發(fā)送；
d）對(duì)于那些未確認(rèn)的消息或者取消的消息，需要有一個(gè)消息狀態(tài)確認(rèn)系統(tǒng)定時(shí)去支付寶系統(tǒng)查詢這個(gè)消息的狀態(tài)并進(jìn)行更新。為什么需要這一步驟，舉個(gè)例子：假設(shè)在第2步支付寶扣款事務(wù)被成功提交后，系統(tǒng)掛了，此時(shí)消息狀態(tài)并未被更新為“確認(rèn)發(fā)送”，從而導(dǎo)致消息不能被發(fā)送。
優(yōu)點(diǎn)：消息數(shù)據(jù)獨(dú)立存儲(chǔ)，降低業(yè)務(wù)系統(tǒng)與消息系統(tǒng)間的耦合；
缺點(diǎn)：一次消息發(fā)送需要兩次請(qǐng)求；業(yè)務(wù)處理服務(wù)需要實(shí)現(xiàn)消息狀態(tài)回查接口。
4、那么如何解決消息重復(fù)投遞的問題？
還有一個(gè)很嚴(yán)重的問題就是消息重復(fù)投遞，以我們支付寶轉(zhuǎn)賬到余額寶為例，如果相同的消息被重復(fù)投遞兩次，那么我們余額寶賬戶將會(huì)增加2萬而不是1萬了(上面講順序消費(fèi)是講過，這里再提一下)。
為什么相同的消息會(huì)被重復(fù)投遞？比如余額寶處理完消息msg后，發(fā)送了處理成功的消息給支付寶，正常情況下支付寶應(yīng)該要?jiǎng)h除消息msg，但如果支付寶這時(shí)候悲劇的掛了，重啟后一看消息msg還在，就會(huì)繼續(xù)發(fā)送消息msg。
解決方法很簡(jiǎn)單，在余額寶這邊增加消息應(yīng)用狀態(tài)表（message_apply），通俗來說就是個(gè)賬本，用于記錄消息的消費(fèi)情況，每次來一個(gè)消息，在真正執(zhí)行之前，先去消息應(yīng)用狀態(tài)表中查詢一遍，如果找到說明是重復(fù)消息，丟棄即可，如果沒找到才執(zhí)行，同時(shí)插入到消息應(yīng)用狀態(tài)表（同一事務(wù)） 。
For each msg in queue
Begin transaction
select count(*) as cnt from message_apply where msg_id=msg.msg_id;
if cnt==0 then
update B set amount=amount+10000 where userId=1;
insert into message_apply(msg_id) values(msg.msg_id);
end if
End transaction
commit;
End For
為了方便大家理解，我們?cè)賮砼e一個(gè)銀行轉(zhuǎn)賬的示例（和上一個(gè)例子差不多）：
比如，Bob向Smith轉(zhuǎn)賬100塊。
在單機(jī)環(huán)境下，執(zhí)行事務(wù)的情況，大概是下面這個(gè)樣子：
當(dāng)用戶增長到一定程度，Bob和Smith的賬戶及余額信息已經(jīng)不在同一臺(tái)服務(wù)器上了，那么上面的流程就變成了這樣：
這時(shí)候你會(huì)發(fā)現(xiàn)，同樣是一個(gè)轉(zhuǎn)賬的業(yè)務(wù)，在集群環(huán)境下，耗時(shí)居然成倍的增長，這顯然是不能夠接受的。那如何來規(guī)避這個(gè)問題？
5、大事務(wù) = 小事務(wù) + 異步
將大事務(wù)拆分成多個(gè)小事務(wù)異步執(zhí)行。這樣基本上能夠?qū)⒖鐧C(jī)事務(wù)的執(zhí)行效率優(yōu)化到與單機(jī)一致。轉(zhuǎn)賬的事務(wù)就可以分解成如下兩個(gè)小事務(wù)：
圖中執(zhí)行本地事務(wù)（Bob賬戶扣款）和發(fā)送異步消息應(yīng)該保證同時(shí)成功或者同時(shí)失敗，也就是扣款成功了，發(fā)送消息一定要成功，如果扣款失敗了，就不能再發(fā)送消息。那問題是：我們是先扣款還是先發(fā)送消息呢？
首先看下先發(fā)送消息的情況，大致的示意圖如下：
存在的問題是：如果消息發(fā)送成功，但是扣款失敗，消費(fèi)端就會(huì)消費(fèi)此消息，進(jìn)而向Smith賬戶加錢。
先發(fā)消息不行，那就先扣款吧，大致的示意圖如下：
存在的問題跟上面類似：如果扣款成功，發(fā)送消息失敗，就會(huì)出現(xiàn)Bob扣錢了，但是Smith賬戶未加錢。
可能大家會(huì)有很多的方法來解決這個(gè)問題，比如：直接將發(fā)消息放到Bob扣款的事務(wù)中去，如果發(fā)送失敗，拋出異常，事務(wù)回滾。這樣的處理方式也符合“恰好”不需要解決的原則。
RocketMQ支持事務(wù)消息，下面來看看RocketMQ是怎樣來實(shí)現(xiàn)的?
RocketMQ第一階段發(fā)送Prepared消息時(shí)，會(huì)拿到消息的地址，第二階段執(zhí)行本地事物，第三階段通過第一階段拿到的地址去訪問消息，并修改消息的狀態(tài)。
細(xì)心的你可能又發(fā)現(xiàn)問題了，如果確認(rèn)消息發(fā)送失敗了怎么辦？RocketMQ會(huì)定期掃描消息集群中的事物消息，如果發(fā)現(xiàn)了Prepared消息，它會(huì)向消息發(fā)送端(生產(chǎn)者)確認(rèn)，Bob的錢到底是減了還是沒減呢？如果減了是回滾還是繼續(xù)發(fā)送確認(rèn)消息呢？
RocketMQ會(huì)根據(jù)發(fā)送端設(shè)置的策略來決定是回滾還是繼續(xù)發(fā)送確認(rèn)消息。這樣就保證了消息發(fā)送與本地事務(wù)同時(shí)成功或同時(shí)失敗。
6、Rocket事務(wù)流程處理分析
那我們來看下RocketMQ源碼，是如何處理事務(wù)消息的。
客戶端發(fā)送事務(wù)消息的部分(完整代碼請(qǐng)查看：rocketmq-example工程下的com.alibaba.rocketmq.example.transaction.TransactionProducer)
// =============================發(fā)送事務(wù)消息的一系列準(zhǔn)備工作========================================
// 未決事務(wù)，MQ服務(wù)器回查客戶端
// 也就是上文所說的，當(dāng)RocketMQ發(fā)現(xiàn)`Prepared消息`時(shí)，會(huì)根據(jù)這個(gè)Listener實(shí)現(xiàn)的策略來決斷事務(wù)
TransactionCheckListener transactionCheckListener = new TransactionCheckListenerImpl();
// 構(gòu)造事務(wù)消息的生產(chǎn)者
TransactionMQProducer producer = new TransactionMQProducer("groupName");
// 設(shè)置事務(wù)決斷處理類
producer.setTransactionCheckListener(transactionCheckListener);
// 本地事務(wù)的處理邏輯，相當(dāng)于示例中檢查Bob賬戶并扣錢的邏輯
TransactionExecuterImpl tranExecuter = new TransactionExecuterImpl();
producer.start()
// 構(gòu)造MSG，省略構(gòu)造參數(shù)
Message msg = new Message(......);
// 發(fā)送消息
SendResult sendResult = producer.sendMessageInTransaction(msg, tranExecuter, null);
producer.shutdown();
接著查看sendMessageInTransaction方法的源碼，總共分為3個(gè)階段：發(fā)送Prepared消息、執(zhí)行本地事務(wù)、發(fā)送確認(rèn)消息。
// ================================事務(wù)消息的發(fā)送過程=============================================
public TransactionSendResult sendMessageInTransaction(.....) {
// 邏輯代碼，非實(shí)際代碼
// 1.發(fā)送消息
sendResult = this.send(msg);
// sendResult.getSendStatus() == SEND_OK
// 2.如果消息發(fā)送成功，處理與消息關(guān)聯(lián)的本地事務(wù)單元
LocalTransactionState localTransactionState = tranExecuter.executeLocalTransactionBranch(msg, arg);
// 3.結(jié)束事務(wù)
this.endTransaction(sendResult, localTransactionState, localException);
}
endTransaction方法會(huì)將請(qǐng)求發(fā)往broker(mq server)去更新事務(wù)消息的最終狀態(tài)：
如果endTransaction方法執(zhí)行失敗，數(shù)據(jù)沒有發(fā)送到broker，導(dǎo)致事務(wù)消息的 狀態(tài)更新失敗，broker會(huì)有回查線程定時(shí)（默認(rèn)1分鐘）掃描每個(gè)存儲(chǔ)事務(wù)狀態(tài)的表格文件，如果是已經(jīng)提交或者回滾的消息直接跳過，如果是prepared狀態(tài)則會(huì)向Producer發(fā)起CheckTransaction請(qǐng)求，Producer會(huì)調(diào)用DefaultMQProducerImpl.checkTransactionState()方法來處理broker的定時(shí)回調(diào)請(qǐng)求，而checkTransactionState會(huì)調(diào)用我們的事務(wù)設(shè)置的決斷方法來決定是回滾事務(wù)還是繼續(xù)執(zhí)行，最后調(diào)用endTransactionOneway讓broker來更新消息的最終狀態(tài)。
再回到轉(zhuǎn)賬的例子，如果Bob的賬戶的余額已經(jīng)減少，且消息已經(jīng)發(fā)送成功，Smith端開始消費(fèi)這條消息，這個(gè)時(shí)候就會(huì)出現(xiàn)消費(fèi)失敗和消費(fèi)超時(shí)兩個(gè)問題，解決超時(shí)問題的思路就是一直重試，直到消費(fèi)端消費(fèi)消息成功，整個(gè)過程中有可能會(huì)出現(xiàn)消息重復(fù)的問題，按照前面的思路解決即可。
消費(fèi)事務(wù)消息
這樣基本上可以解決消費(fèi)端超時(shí)問題，但是如果消費(fèi)失敗怎么辦？阿里提供給我們的解決方法是：人工解決。大家可以考慮一下，按照事務(wù)的流程，因?yàn)槟撤N原因Smith加款失敗，那么需要回滾整個(gè)流程。如果消息系統(tǒng)要實(shí)現(xiàn)這個(gè)回滾流程的話，系統(tǒng)復(fù)雜度將大大提升，且很容易出現(xiàn)Bug，估計(jì)出現(xiàn)Bug的概率會(huì)比消費(fèi)失敗的概率大很多。這也是RocketMQ目前暫時(shí)沒有解決這個(gè)問題的原因，在設(shè)計(jì)實(shí)現(xiàn)消息系統(tǒng)時(shí)，我們需要衡量是否值得花這么大的代價(jià)來解決這樣一個(gè)出現(xiàn)概率非常小的問題，這也是大家在解決疑難問題時(shí)需要多多思考的地方。
我們需要注意的是，在3.2.6版本中移除了事務(wù)消息的實(shí)現(xiàn)，所以此版本不支持事務(wù)消息。也就是說，消息失敗不會(huì)進(jìn)行檢查。
7、交易事務(wù)處理示例
下面我們來看一個(gè)簡(jiǎn)單的例子：
消息生產(chǎn)者：
public class TransactionProducer {
public static void main(String[] args) throws MQClientException, InterruptedException {
TransactionListener transactionListener = new TransactionListenerImpl();
TransactionMQProducer producer = new TransactionMQProducer("transaction_producer");
//設(shè)置用于事務(wù)消息的處理線程池
ExecutorService executorService = new ThreadPoolExecutor(2, 5, 100, TimeUnit.SECONDS, new ArrayBlockingQueue<Runnable>(2000), new ThreadFactory() {
@Override
public Thread newThread(Runnable r) {
Thread thread = new Thread(r);
thread.setName("client-transaction-msg-check-thread");
return thread;
}
});
producer.setNamesrvAddr("127.0.0.1:9876");
producer.setExecutorService(executorService);
//設(shè)置事務(wù)監(jiān)聽器，監(jiān)聽器實(shí)現(xiàn)接口org.apache.rocketmq.client.producer.TransactionListener
//監(jiān)聽器中實(shí)現(xiàn)需要處理的交易業(yè)務(wù)邏輯的處理，以及MQ Broker中未確認(rèn)的事務(wù)與業(yè)務(wù)的確認(rèn)邏輯
producer.setTransactionListener(transactionListener);
producer.start();
//生成不同的Tag，用于模擬不同的處理場(chǎng)景
String[] tags = new String[] {"TagA", "TagB", "TagC", "TagD", "TagE"};
for (int i = 0; i < 10; i++) {
try {
//組裝產(chǎn)生消息
Message msg =
new Message("TopicTransaction", tags[i % tags.length], "KEY" + i,
("Hello RocketMQ " + i).getBytes(RemotingHelper.DEFAULT_CHARSET));
//以事務(wù)發(fā)送消息，并在事務(wù)消息被成功預(yù)寫入到RocketMQ中后，執(zhí)行用戶定義的交易邏輯，
//交易邏輯執(zhí)行成功后，再實(shí)現(xiàn)實(shí)現(xiàn)業(yè)務(wù)消息的提交邏輯
SendResult sendResult = producer.sendMessageInTransaction(msg, null);
System.out.printf("%s%n", sendResult);
System.out.printf("%s%n", sendResult.getTransactionId());
Thread.sleep(10);
} catch (MQClientException | UnsupportedEncodingException e) {
e.printStackTrace();
}
}
producer.shutdown();
}
}
業(yè)務(wù)實(shí)現(xiàn)類TransactionListenerImpl：
public class TransactionListenerImpl implements TransactionListener {
private AtomicInteger transactionIndex = new AtomicInteger(0);
private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<>();
/**
* 該方法會(huì)在消息成功預(yù)寫入RocketMQ后被執(zhí)行
*/
@Override
public LocalTransactionState executeLocalTransaction(Message msg, Object arg) {
int value = transactionIndex.getAndIncrement();
System.out.println("開始處理業(yè)務(wù)邏輯...");
int status = value % 3;
localTrans.put(msg.getTransactionId(), status);
switch (status) {
case 0:
//LocalTransactionState.UNKNOW表示未知的事件，需要RocketMQ進(jìn)一步服務(wù)業(yè)務(wù)進(jìn)行確認(rèn)該交易的處理
//結(jié)果，確認(rèn)消息被調(diào)用的方法為下方的checkLocalTransaction。
//注：RocketMQ與業(yè)務(wù)確認(rèn)消息的執(zhí)行狀態(tài)的功能已經(jīng)被移除了，在早期3.0.8的版本中有該功能，因而如果
//返回的狀態(tài)為UNKNOW，則該消息不會(huì)被提交
return LocalTransactionState.UNKNOW;
case 1:
return LocalTransactionState.COMMIT_MESSAGE;
case 2:
return LocalTransactionState.ROLLBACK_MESSAGE;
default:
return LocalTransactionState.COMMIT_MESSAGE;
}
}
/**
* 該方法用于RocketMQ與業(yè)務(wù)確認(rèn)未提交事務(wù)的消息的狀態(tài)，不過該方法已經(jīng)的實(shí)現(xiàn)在RocketMQ中已經(jīng)
* 被刪除了，因而其功能也就沒有意義了。
* 不過如果使用阿里云的企業(yè)的RocketMQ服務(wù)，該功能會(huì)起作用。
*/
@Override
public LocalTransactionState checkLocalTransaction(MessageExt msg) {
Integer status = localTrans.get(msg.getTransactionId());
int mod = msg.getTransactionId().hashCode() % 2;
if (null != status) {
switch (mod) {
case 0:
return LocalTransactionState.ROLLBACK_MESSAGE;
case 1:
return LocalTransactionState.COMMIT_MESSAGE;
default:
return LocalTransactionState.COMMIT_MESSAGE;
}
}
return LocalTransactionState.COMMIT_MESSAGE;
}
}
消息消費(fèi)者：
public class Consumer {
public static void main(String[] args) throws MQClientException {
DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("transaction_producer");
consumer.setNamesrvAddr("127.0.0.1:9876");
/**
* 設(shè)置Consumer第一次啟動(dòng)是從隊(duì)列頭部開始消費(fèi)還是隊(duì)列尾部開始消費(fèi)<br>
* 如果非第一次啟動(dòng)，那么按照上次消費(fèi)的位置繼續(xù)消費(fèi)
*/
consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);
consumer.subscribe("TopicTransaction", "*");
consumer.registerMessageListener(new MessageListenerOrderly() {
private Random random = new Random();
@Override
public ConsumeOrderlyStatus consumeMessage(List<MessageExt> msgs, ConsumeOrderlyContext context) {
// 設(shè)置自動(dòng)提交
context.setAutoCommit(true);
for (MessageExt msg : msgs) {
System.out.println("獲取到消息開始消費(fèi)："+msg + " , content : " + new String(msg.getBody()));
}
try {
// 模擬業(yè)務(wù)處理
TimeUnit.SECONDS.sleep(random.nextInt(5));
} catch (Exception e) {
e.printStackTrace();
//返回處理失敗，該消息后續(xù)可以繼續(xù)被消費(fèi)
return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
}
//返回處理成功，該消息就不會(huì)再次投遞過來了
return ConsumeOrderlyStatus.SUCCESS;
}
});
consumer.start();
System.out.println("consumer start ! ");
}
}
我們先啟動(dòng)消費(fèi)端，然后啟動(dòng)生產(chǎn)端：
在運(yùn)行之前，我們先來看一下，web控制臺(tái)的消息：
Topic transaction_producer中沒有未消費(fèi)的消息，下面開始執(zhí)行代碼邏輯。
生產(chǎn)端：
通過日志可以看到其發(fā)送了四條消息，交易邏輯被調(diào)用了4次，其中只有一條消息反饋的結(jié)果為COMMIT_MESSAGE。
再次查看控制臺(tái)：
可以看到其Topic transaction_producer只有一條待消費(fèi)的消息，這個(gè)和發(fā)送端只一條消息被COMMIT的結(jié)論相符合。
消費(fèi)端：
啟動(dòng)消費(fèi)端，在控制臺(tái)只看到消費(fèi)了一條消息。
生產(chǎn)者總共生產(chǎn)了四條消息，原因如下：
這就是為什么我們生產(chǎn)了四條消息，最后卻只消費(fèi)了一條，再次確認(rèn)業(yè)務(wù)實(shí)現(xiàn)類TransactionListenerImpl中的方法checkLocalTransaction被沒有調(diào)用。
8、如何保證扣錢與加錢的事務(wù)的最終一致性
在上面的轉(zhuǎn)賬交易邏輯中，存在兩個(gè)問題：
1）如果沒有使用RocketMQ的企業(yè)版本，那就可能會(huì)發(fā)生扣錢的事務(wù)成功了，但是扣錢的消息由于生產(chǎn)方發(fā)生了故障，導(dǎo)致交易消息沒有在扣錢的事務(wù)提交成功后往RocketMQ中確認(rèn)該條消息可以被提交，就會(huì)導(dǎo)致該條消息不會(huì)遞交給消費(fèi)方，導(dǎo)致Bob的錢被扣了，但是Smith的錢卻沒有增加。
2）生產(chǎn)方的全部邏輯都處理完成了，扣錢的事務(wù)在數(shù)據(jù)庫中被成功的提交，扣錢的消息在RocketMQ被成功的確認(rèn)，但是消費(fèi)方在消費(fèi)消息的時(shí)候，自己本身發(fā)生了故障，或者處理該條消息發(fā)生了邏輯錯(cuò)誤，導(dǎo)致Smith的錢沒有被正確的加上。
以上兩個(gè)問題雖然發(fā)生的機(jī)率都很低，但是只要存在著發(fā)生的機(jī)率就會(huì)一定在某個(gè)時(shí)間點(diǎn)發(fā)生，只是故障發(fā)生時(shí)間點(diǎn)的早晚問題。在金融系統(tǒng)中，每日都會(huì)跑系統(tǒng)日志執(zhí)行對(duì)賬操作，用于核對(duì)當(dāng)日總共的支付與收入是否是平衡的、每個(gè)單筆交易結(jié)果是否都滿足借貸平衡等，因而為了避免以上兩個(gè)問題的發(fā)生，我的處理方式還是引入金融系統(tǒng)對(duì)賬的業(yè)務(wù)邏輯來進(jìn)行處理。
其業(yè)務(wù)處理邏輯如下：
在發(fā)送交易事務(wù)消息過后，發(fā)送一個(gè)交易對(duì)賬消息到對(duì)賬Topic中，該對(duì)賬消息為非事務(wù)消息，發(fā)送成功即表示成功保存到了RocketMQ中，該交易對(duì)賬消息不會(huì)用于消費(fèi)者消費(fèi)，后續(xù)的交易對(duì)賬系統(tǒng)會(huì)消費(fèi)該隊(duì)列中的對(duì)賬信息，其分別會(huì)和交易的生產(chǎn)方和消費(fèi)方進(jìn)行交易核對(duì)，核對(duì)邏輯如下：
交易對(duì)賬系統(tǒng)首先和交易的消費(fèi)方進(jìn)行核對(duì)，如果消費(fèi)方消費(fèi)成功，則可以說明整個(gè)交易結(jié)果滿足最終一致性，因?yàn)橄⑹巧a(chǎn)者成功處理后，然后再發(fā)送的交易確認(rèn)消息，因而只要產(chǎn)生了事務(wù)確認(rèn)的交易消息，則可以肯定生產(chǎn)方已經(jīng)正常執(zhí)行完了扣款的邏輯。
只有交易消息在消費(fèi)方處理失敗或者消息方?jīng)]有消費(fèi)該消息的情況下，才需要再次和生產(chǎn)方進(jìn)行確認(rèn)，如果生產(chǎn)方成功執(zhí)行了扣款操作，則需要回滾這筆扣款交易；如果沒有扣除成功，則表示兩邊都沒有消費(fèi)這邊交易，就不用做任何操作了。
版權(quán)聲明：本文為博主原創(chuàng)文章，遵循 CC 4.0 BY-SA 版權(quán)協(xié)議，轉(zhuǎn)載請(qǐng)附上原文出處鏈接和本聲明。
本文鏈接：
https://blog.csdn.net/fenglibing/article/details/92417739
