金融項(xiàng)目-課件分布式事務(wù)

1.什么是事務(wù)？舉個(gè)生活中的例子：你去商店買東西就是一個(gè)事務(wù)的例子，買東西是一個(gè)交易，包含“手交錢，一手交貨”1.什么是事務(wù)？舉個(gè)生活中的例子：你去商店買東西就是一個(gè)事務(wù)的例子，買東西是一個(gè)交易，包含“手交錢，一手交貨”作失敗，交易就必須撤銷。系型數(shù)據(jù)庫(kù)的事務(wù)又被稱為本地事務(wù)?；仡櫼幌聰?shù)據(jù)庫(kù)事務(wù)的四大特性A（Atomic）功部分失敗的情況。C（Consistency轉(zhuǎn)100元，轉(zhuǎn)賬前和轉(zhuǎn)賬后的數(shù)據(jù)是正確狀態(tài)這叫一致性，如果出現(xiàn)張三轉(zhuǎn)出100元，李四賬戶沒(méi)有增加100元這就出現(xiàn)了數(shù)據(jù)錯(cuò)誤，就沒(méi)有達(dá)到一致性。I（Isolatio）：隔離性，數(shù)據(jù)庫(kù)中的事務(wù)一般都是并發(fā)的，隔離性是指并發(fā)的兩個(gè)事務(wù)的執(zhí)行互不干題。D（Duability滾。微服務(wù)應(yīng)用的演變：成事務(wù)操作，這種分布式系統(tǒng)環(huán)境下的事務(wù)機(jī)制稱之為分布式事務(wù)。begin--1.--2.成事務(wù)操作，這種分布式系統(tǒng)環(huán)境下的事務(wù)機(jī)制稱之為分布式事務(wù)。begin--1.--2.本地?cái)?shù)據(jù)庫(kù)操作：李四增加金額committransaction;begin--1.A--2.A微服務(wù)遠(yuǎn)程調(diào)用B微服務(wù)：讓李四增加金額committransaction;因此在分布式架構(gòu)的基礎(chǔ)上，傳統(tǒng)數(shù)據(jù)庫(kù)事務(wù)就無(wú)法使用了，比如上例，張三和李四的賬戶不在一個(gè)數(shù)據(jù)庫(kù)中甚至不在一個(gè)應(yīng)用系統(tǒng)里，怎么實(shí)現(xiàn)轉(zhuǎn)賬事務(wù)？也就是說(shuō)同樣一個(gè)功能，原來(lái)是由一個(gè)系統(tǒng)完成的，即使這個(gè)功能包含很多個(gè)操作，也可以采用數(shù)據(jù)庫(kù)事務(wù)(本地事務(wù))搞定，而現(xiàn)在這個(gè)功能中包含的多個(gè)操作可能是由多個(gè)系統(tǒng)(微服務(wù))參與完成的，此時(shí)數(shù)據(jù)庫(kù)事務(wù)(本地事務(wù))的分布式事務(wù)理論來(lái)支撐了。從而幫助我們理解每個(gè)解決方案。CAP理解求，就需要橫向去擴(kuò)展幾臺(tái)Slave(從數(shù)據(jù)庫(kù))去分擔(dān)求，就需要橫向去擴(kuò)展幾臺(tái)Slave(從數(shù)據(jù)庫(kù))去分擔(dān)Master(主數(shù)據(jù)庫(kù))的壓力。如果服務(wù)對(duì)數(shù)據(jù)庫(kù)的需求是IO密集型的，那可能會(huì)經(jīng)常遇到增刪改影響到了查詢效率。這就需要同步。1、商品服務(wù)請(qǐng)求主數(shù)據(jù)庫(kù)寫入商品信息（添加商品、修改商品、刪除商品23C-到的數(shù)據(jù)都是最新的狀態(tài)。12A-Availability上圖中，商品信息讀取滿足可用性就是要實(shí)現(xiàn)如下目標(biāo)：12P-Partitiontolerance部分還能維持分布式系統(tǒng)的運(yùn)作，這樣就具有較好的分區(qū)容忍性。1部分還能維持分布式系統(tǒng)的運(yùn)作，這樣就具有較好的分區(qū)容忍性。122.1.2CAP1、上邊的例子是否同時(shí)具備CAP在所有分布式事務(wù)場(chǎng)景中不會(huì)同時(shí)具備CAP三個(gè)特性，因?yàn)樵诰邆淞薖的前提下C和A如果要保證數(shù)據(jù)的一致性就要實(shí)現(xiàn)每個(gè)節(jié)點(diǎn)的數(shù)據(jù)一致，節(jié)點(diǎn)越多可用性越好，但是數(shù)據(jù)一致性會(huì)越差。2、CAP例如：上邊的商品管理，完全可以實(shí)現(xiàn)AP最新的即可。通常實(shí)現(xiàn)AP都會(huì)保證最終一致性，后面講的ASE理論就是根據(jù)AP來(lái)擴(kuò)展的，一些業(yè)務(wù)場(chǎng)景比如：訂單退款，今日退款成功，明日賬戶到賬，只要用戶可以接受在一定時(shí)間內(nèi)到賬即可。放棄可用性，追求一致性和分區(qū)容錯(cuò)性，我們的zooeeper一次轉(zhuǎn)賬請(qǐng)求要等待雙方銀行系統(tǒng)都完成整個(gè)事務(wù)才算完成。性。那么系統(tǒng)將不是一個(gè)標(biāo)準(zhǔn)的分布式系統(tǒng)，我們最常用的關(guān)系型數(shù)據(jù)庫(kù)就滿足了CA。上邊的商品管理，如果要實(shí)現(xiàn)CA2.1.3通過(guò)上面我們已經(jīng)學(xué)習(xí)了CAP理論的相關(guān)知識(shí)，CAP是一個(gè)已經(jīng)被證實(shí)的理論：一個(gè)分布式系統(tǒng)最多只能同時(shí)滿足一致性（Consistency）、可用性（Availability2.1.3通過(guò)上面我們已經(jīng)學(xué)習(xí)了CAP理論的相關(guān)知識(shí)，CAP是一個(gè)已經(jīng)被證實(shí)的理論：一個(gè)分布式系統(tǒng)最多只能同時(shí)滿足一致性（Consistency）、可用性（Availability）和分區(qū)容忍性（Partitiontoleance）這三項(xiàng)中的兩項(xiàng)。它可以作為我們進(jìn)行架構(gòu)設(shè)計(jì)、技術(shù)選型的考量標(biāo)準(zhǔn)。對(duì)于多數(shù)大型互聯(lián)網(wǎng)應(yīng)用的場(chǎng)景，節(jié)點(diǎn)眾多、部署分散，而且現(xiàn)在的集群規(guī)模越來(lái)越大，所以節(jié)點(diǎn)故障、網(wǎng)絡(luò)故障是常態(tài)，而且要保證服務(wù)可用性達(dá)到N個(gè)9（99.99..%），擇：保證P和A，舍棄C強(qiáng)一致，保證最終一致性。2.2BASE1CAP理論告訴我們一個(gè)分布式系統(tǒng)最多只能同時(shí)滿足一致性（Consistency）、可用性（Availability）和分區(qū)容忍性（Partitiontoleance）這三項(xiàng)中的兩項(xiàng)，其中AP在實(shí)際應(yīng)用中較多，AP保證可用性和分區(qū)容忍性，但是在實(shí)際生產(chǎn)中很多場(chǎng)景都要實(shí)現(xiàn)一致性，比如前邊我們舉的例子，主數(shù)據(jù)庫(kù)向從數(shù)據(jù)庫(kù)同步數(shù)據(jù)，即使不要一致性，但是最終也要將數(shù)據(jù)同步成功來(lái)保證數(shù)據(jù)一致，這種一致性和CAP中的一致性不同，CAP中的一致性要求在任何時(shí)間查詢每個(gè)節(jié)點(diǎn)數(shù)據(jù)都必須一致，它強(qiáng)調(diào)的是強(qiáng)一致性，但是最終一致性是允許可以在一段時(shí)間內(nèi)每個(gè)節(jié)點(diǎn)的數(shù)據(jù)不一致，但是經(jīng)過(guò)一段時(shí)間每個(gè)節(jié)點(diǎn)的數(shù)據(jù)必須一致，它強(qiáng)調(diào)的是最終數(shù)據(jù)的一致性。2、BaseASE是BasicallyAvailable(基本可用)、Softstate(軟狀態(tài))和Eventuallyconsistent(最終一致性)短語(yǔ)的縮寫。ASE理論是對(duì)CAP中AP的一個(gè)擴(kuò)展，通過(guò)犧牲強(qiáng)一致性來(lái)獲得可用性，當(dāng)出現(xiàn)故障允許部分不可用但要保證核心功能可用，允許數(shù)據(jù)在一段時(shí)間內(nèi)是不一致的，但最終達(dá)到一致?tīng)顟B(tài)。滿足ASE理論的事務(wù)，我們稱之為柔性事務(wù)”?；究捎?分布式系統(tǒng)在出現(xiàn)故障時(shí)，允許損失部分可用功能，保證核心功能可用。如，電商網(wǎng)站軟狀態(tài):由于不要求強(qiáng)一致性，所以ASE允許系統(tǒng)中存在中間狀態(tài)（軟狀態(tài)），這個(gè)狀態(tài)不影響系統(tǒng)可用性，如訂單的"支付中"、“數(shù)據(jù)同步中”等狀態(tài)，待數(shù)據(jù)最終一致后狀態(tài)改為“成功”態(tài)。最終一致:最終一致是指經(jīng)過(guò)一段時(shí)間后，所有節(jié)點(diǎn)數(shù)據(jù)都將會(huì)達(dá)到一致。如訂單的"支付中"狀態(tài)，最終會(huì)變?yōu)橹Ц冻晒Α被蛘?支付失敗"間的延遲、等待。TCCC(ry/Con?rm/Cancel)編程模式的核心思想是：針對(duì)每個(gè)分支事務(wù)操作，都要向全局事務(wù)發(fā)起方注冊(cè)ry、Con?rm和Cancel執(zhí)行：Try階段主要是做業(yè)務(wù)檢查(一致性)及資源預(yù)留(隔離)Con?rmCon?rm階段主要是做確認(rèn)提交，Try階段所有分支事務(wù)執(zhí)行成功后開始執(zhí)行Con?rm。通常情況下，采用TCC則認(rèn)為Con?rm階段是不會(huì)出錯(cuò)的。即：只要Try成功，Con?rm一定成功。若Cancel放。通常情況下，采用TCC則認(rèn)為Cancel階段也是一定成功的。若Canceltry-con?rm-cancel中，由事務(wù)管理器(協(xié)調(diào)管理)進(jìn)AB兩個(gè)try分別執(zhí)行，在這個(gè)過(guò)程中，事務(wù)管理器會(huì)對(duì)AB進(jìn)行監(jiān)控，一旦任何一方出現(xiàn)了問(wèn)題，就推進(jìn)對(duì)方執(zhí)行cancel；如果雙方都沒(méi)有異常，就推進(jìn)AB執(zhí)行con?rm。如果在執(zhí)行con?rm或cancel過(guò)程中出現(xiàn)問(wèn)題，就引入重試機(jī)制或由人工處理。C解決方案要求每個(gè)分支事務(wù)實(shí)現(xiàn)三個(gè)操作ry/Con?rm/Cancel。try操作做業(yè)務(wù)檢查及資源預(yù)留，Con?rm操作做業(yè)務(wù)確認(rèn)操作，Cancel操作需要實(shí)現(xiàn)一個(gè)與try相反的操作。TM(事務(wù)管理器)有的分支事務(wù)的try操作，任何一個(gè)分支事務(wù)的try操作執(zhí)行失敗，TM將會(huì)發(fā)起所有分支事務(wù)的Cancel操作，若try操作全部成功，TM將會(huì)發(fā)起所有分支事務(wù)的Con?rm操作，其中Con?rm/Cancel操作若執(zhí)行失敗，TM會(huì)進(jìn)行重試，因此需要實(shí)現(xiàn)冪等。ry/Con?rm/Cancel據(jù)業(yè)務(wù)情況靈活掌握。對(duì)應(yīng)用的侵入性強(qiáng)。業(yè)務(wù)邏輯的每個(gè)分支都需要實(shí)現(xiàn)try、con?rm、cancel較強(qiáng)，改造成本高。致性的要求，con?rm和cancel接口必須實(shí)現(xiàn)冪等。3.2TCC失敗，TM會(huì)進(jìn)行重試，因此需要實(shí)現(xiàn)冪等。ry/Con?rm/Cancel據(jù)業(yè)務(wù)情況靈活掌握。對(duì)應(yīng)用的侵入性強(qiáng)。業(yè)務(wù)邏輯的每個(gè)分支都需要實(shí)現(xiàn)try、con?rm、cancel較強(qiáng)，改造成本高。致性的要求，con?rm和cancel接口必須實(shí)現(xiàn)冪等。3.2TCC目前市面上的TCC支持嵌套事務(wù)(Nestedtransactionsupport)采用disruptor框架進(jìn)行事務(wù)日志的異步讀寫，與RPC支持SpringBoot-starter項(xiàng)目啟動(dòng)，使用簡(jiǎn)單。RPC框架支持:dubbo,motan,springcloud本地事務(wù)存儲(chǔ)支持:redis,mongodb,zookeeper,?le,mysql。事務(wù)日志序列化支持：java，hessian，kryo，protostu?。采用AspectAOP切面思想與Spring無(wú)縫集成，天然支持集群。Hmily利用OP透明的調(diào)用到另一方的ry、Conform、Cancel償，重試等。Hmily不需要事務(wù)協(xié)調(diào)服務(wù)，但需要提供一個(gè)數(shù)據(jù)庫(kù)(mysql/mongodb/zooeeper/edis/?le)來(lái)進(jìn)行日志存儲(chǔ)。Hmily實(shí)現(xiàn)的C服務(wù)與普通的服務(wù)一樣，只需要暴露一個(gè)接口，也就是它的ry業(yè)務(wù)。Con?rm/Cancel業(yè)務(wù)邏輯，只是因?yàn)槿质聞?wù)提交/回滾的需要才提供的，因此Con?rm/Cancel需要被Hmily事務(wù)框架發(fā)現(xiàn)即可，不需要被調(diào)用它的其他業(yè)務(wù)服務(wù)所感知。官網(wǎng)介紹：/website/zh-本案例通過(guò)hmily框架實(shí)現(xiàn)C分布式事務(wù)，模擬兩個(gè)賬戶的轉(zhuǎn)賬交易過(guò)程。兩個(gè)賬戶分別在不同的銀行(張三在bank、李四在bank2)，bank1、bank2成功，要么都失敗。tcc-/changmingxie/tcc-/QNJR-數(shù)據(jù)庫(kù)：MySQL5.7.25+JDK：jdk1.8+微服務(wù)：數(shù)據(jù)庫(kù)：MySQL5.7.25+JDK：jdk1.8+微服務(wù)：spring-boot-2.1.3、spring-cloud-hmily：hmily-springcloud.2.0.4-創(chuàng)建bank1CREATEDATABASE`bank1`CHARACTERSET'utf8'COLLATEUSEDROPTABLEIFEXISTS`account_info`;CREATETABLEUSEDROPTABLEIFEXISTS`account_info`;CREATETABLE`account_info`(`id`bigint(20)NOTNULL`account_name`varchar(100CHARACTERSETutf8COLLATEutf8_binNULLDEFAULTNULLCOMMENT'戶主姓名',`account_no`varchar(100CHARACTERSETutf8COLLATEutf8_binNULLDEFAULTNULLCOMMENT'銀行卡號(hào)',`account_password`varchar(100CHARACTERSETutf8COLLATEutf8_binNULLDEFAULTNULLCOMMENT'帳戶密碼',`account_balance`doubleNULLDEFAULTNULLCOMMENT'帳戶余額PRIMARYKEY(`id`)USING)ENGINE=InnoDBAUTO_INCREMENT=5CHARACTERSET=utf8COLLATE=ROW_FORMAT=INSERTINTO`account_info`VALUES(1'張三''1'''創(chuàng)建bank2庫(kù)，并導(dǎo)入以下表結(jié)構(gòu)和數(shù)據(jù)CREATEDATABASE`bank2`CHARACTERSET'utf8'COLLATEUSEDROPTABLEIFEXISTS`account_info`;CREATETABLE`account_info`(`id`bigint(20)NOTNULL`account_name`varchar(100CHARACTERSETutf8COLLATEutf8_binNULLDEFAULTNULLCOMMENT'戶主姓名',`account_no`varchar(100CHARACTERSETutf8COLLATEutf8_binNULLDEFAULTNULLCOMMENT'銀行卡號(hào)',`account_password`varchar(100CHARACTERSETutf8COLLATEutf8_binNULLDEFAULTNULLCOMMENT'帳戶密碼',`account_balance`doubleNULLDEFAULTNULLCOMMENT'帳戶余額PRIMARYKEY(`id`)USING)ENGINE=InnoDBAUTO_INCREMENT=5CHARACTERSET=utf8COLLATE=ROW_FORMAT=INSERTINTO`account_info`VALUES(2'李四''2'NULLHmilyMavenpom.xml導(dǎo)入hmily（1）application.yml配置（只顯示hmily部分hmily:serializerkryo#retryMax:hmily:serializerkryo#retryMax:30#最大重試次數(shù)repositorySupportdb#持久化方式started:true#事務(wù)發(fā)起方hmilyDbConfig:driverClassName:username:rootpassword:（2）HmilyprivateEnvironmentpublicHmilyTransactionBootstraphmilyTransactionBootstrap(HmilyInitServiceHmilyTransactionBootstraphmilyTransactionBootstrap=HmilyDbConfighmilyDbConfig=newreturnhmilyTransactionBootstrap;（3）feign@FeignClient(value="hmily-demo-bank2")publicinterfaceBank2Client{（3）feign@FeignClient(value="hmily-demo-bank2")publicinterfaceBank2Client{Booleantransfer(@RequestParam("amount")Double（4）privateBank2Client@Hmily(confirmMethod="commit",cancelMethod="rollback")publicvoidprepare(StringaccountNo,doubleamount){System.out.println("...Bank1Serviceprepare...");thrownewRuntimeException("bank2publicvoidcommit(StringaccountNo,doubleamount){System.out.println("...Bank1Servicecommit..."publicvoidrollback(StringaccountNo,doubleamount){accountInfoDao.updateAccountBalance(accountNo,amountSystem.out.println("...Bank1Servicerollback...");（5）@SpringBootApplication(exclude=MongoAutoConfiguration.class)@EnableFeignClients(basePackagespublicclassBank1HmilyServer{publicstaticvoidmain(String[]args){SpringApplication.run(Bank1HmilyServer.class,args);hmily-demo-bank2（1）application.yml配置（只顯示hmily部分hmily-demo-bank2（1）application.yml配置（只顯示hmily部分hmily:serializerkryo#retryMax:30#最大重試次數(shù)repositorySupportdb#持久化方式started:false#事務(wù)參與方hmilyDbConfig:driverClassName:username:rootpassword:Hmily配置類，和hmily-demo-bank1@Hmily(confirmMethod="confirmMethod",cancelMethod=publicBooleanupdateAccountBalance(StringaccountNo,Doubleamount)System.out.println("...Bank2ServiceBegin...");accountInfoDao.updateAccountBalance(accountNo}catch(Exceptione){thrownewRuntimeException(e.getMessage()returnpublicBooleanconfirmMethod(StringaccountNo,Doubleamount){System.out.println("...Bank2Servicecommit...");returnpublicBooleancancelMethod(StringaccountNo,Doubleamount){accountInfoDao.updateAccountBalance(accountNo,amount*-1);System.out.println("...Bank2Servicerollback...");return（4）@SpringBootApplication(exclude=MongoAutoConfiguration.class)publicclassBank2HmilyServer{@SpringBootApplication(exclude=MongoAutoConfiguration.class)publicclassBank2HmilyServer{publicstaticvoidmain(String[]args)3.3.5對(duì)應(yīng)的賬號(hào)信息，該業(yè)務(wù)存在分布式事務(wù)問(wèn)題。且屬于執(zhí)行時(shí)間較短的業(yè)務(wù)。C方案的軟狀態(tài)時(shí)間很短，一致性較強(qiáng)，因此在此業(yè)務(wù)，我們選用C型分布式事務(wù)解決方案。4.2Hmily1.新建數(shù)據(jù)庫(kù)p2p_undo_log，此庫(kù)用來(lái)存儲(chǔ)hmily事務(wù)日志，空庫(kù)即可，由hmilyCREATEDATABASE`p2p_undo_log`CHARACTERSET'utf8'COLLATE2Maven<artifactId>hmily-<artifactId>hmily-<version>2.0.5-<artifactId>spring-boot-<artifactId>spring-boot-starter-<artifactId>logback-<artifactId>slf4j-<artifactId>logback-<artifactId>spring-(1)在Apollo中，為common-template項(xiàng)目新建公共namespace，命名為micro_service.spring-cloud-org.dromara.hmily.serializer=kryoorg.dromara.hmily.retryMax=30org.dromara.hmily.repositorySupport=dborg.dromara.hmily.started=trueorg.dromara.hmily.hmilyDbConfig.driverClassName=com.mysql.cj.jdbc.Driverorg.dromara.hmily.hmilyDbConfig.url=jdbc:mysql://localhost:3306/p2p_undo_log?org.dromara.hmily.hmilyDbConfig.username=org.dromara.hmily.hmilyDbConfig.password=在consumer-service項(xiàng)目中，關(guān)聯(lián)剛才新建的micro_service.spring-cloud-在application.yml中增加micro_service.spring-cloud-在con?g包中新增HmilypublicclassHmilyConfig在application.yml中增加micro_service.spring-cloud-在con?g包中新增HmilypublicclassHmilyConfig{privateEnvironmentpublicHmilyTransactionBootstraphmilyTransactionBootstrap(HmilyInitServiceHmilyTransactionBootstraphmilyTransactionBootstrap=HmilyDbConfighmilyDbConfig=newreturnhmilyTransactionBootstrap;(4)啟動(dòng)類上增加org.dromara.hmily@SpringBootApplication(scanBasePackages=(5)Feign代理(AccountApiAgent接口)中增加@HmilypublicinterfaceAccountApiAgent{@PostMapping(value=publicinterfaceAccountApiAgent{@PostMapping(value="/account/l/accounts")RestResponse<AccountDTO>register(@RequestBodyAccountRegisterDTO(6)修改ConsumerServiceImpl代碼，注冊(cè)Try、Con?rm、publicvoidregister(ConsumerRegisterDTOconsumerRegisterDTO){...("executeconfirmRegister");("executecancelRegister");4.3.2事務(wù)參與方：account-在Apollo中，為account-service項(xiàng)目關(guān)聯(lián)micro_service.spring-cloud-hmilyorg.dromara.hmily.started=false在application.yml中增加micro_service.spring-cloud-在con?g包中新增Hmily配置類，同consumer-service啟動(dòng)類上增加org.dromara.hmily@SpringBootApplication(scanBasePackages(5)修改AccountServiceImpl代碼，注冊(cè)Try、Con?rm、publicAccountDTOregister(AccountRegisterDTOregisterDTO){...("executeconfirmRegister");("executecancelRegister");//前面在學(xué)習(xí)CAP和ASE理論時(shí)，我們得出結(jié)論：一般情況下會(huì)保證P和A前面在學(xué)習(xí)CAP和ASE理論時(shí)，我們得出結(jié)論：一般情況下會(huì)保證P和A，舍棄C，保證最終一致性。最終一致是指經(jīng)過(guò)一段時(shí)間后，所有節(jié)點(diǎn)數(shù)據(jù)都將會(huì)達(dá)到一致。如訂單的"支付中"狀態(tài)，最終會(huì)變?yōu)椤俺晒Α被蛘?支付失敗"，使訂單狀態(tài)與實(shí)際交易結(jié)果達(dá)成一致，但需要一定時(shí)間的延遲、等待。5.1現(xiàn)最終一致性。以轉(zhuǎn)賬為例：消息發(fā)送方張三，即扣減余額30元，然后通過(guò)網(wǎng)絡(luò)發(fā)送消息到消息接收方李四，即通過(guò)網(wǎng)絡(luò)從MQ中接收消息，然后增加余額301.發(fā)送成功2.commit異常，數(shù)據(jù)庫(kù)回滾，但此時(shí)消息已經(jīng)正常發(fā)送了，同樣會(huì)導(dǎo)致不一致。行成功分丟失，從而導(dǎo)致消息重復(fù)消費(fèi)。5.2RocketMQRocketMQ分丟失，從而導(dǎo)致消息重復(fù)消費(fèi)。5.2RocketMQRocketMQ是一個(gè)來(lái)自阿里巴巴的分布式消息中間件，于2012年開源，并在2017年正式成為Apache頂級(jí)項(xiàng)目。ApacheRocketMQ4.3之后的版本正式支持事務(wù)消息，為分布式事務(wù)實(shí)現(xiàn)提供了便1.發(fā)送成功RocetMQ中的Boer與發(fā)送方具備雙向通信能力，使得boer并且RocetMQ本身提供了存儲(chǔ)機(jī)制，使得事務(wù)消息可以持久化保存；這些優(yōu)秀的設(shè)計(jì)可以保證即使發(fā)生了異常，RocetMQ依然能夠保證達(dá)成事務(wù)的最終一致性。發(fā)送方發(fā)送一個(gè)事務(wù)消息給Boe，RocetM會(huì)將消息狀態(tài)標(biāo)記為“Pepaed時(shí)不能被接收方消費(fèi)。這樣的消息稱之為HalfMessag，即半消息。Broker若本地事務(wù)執(zhí)行成功，發(fā)送commi消息給Boe，RocetMQ會(huì)將消息狀態(tài)標(biāo)記為“可消費(fèi)”，此ollbac消息給將刪除該消息。如果發(fā)送方在本地事務(wù)過(guò)程中，出現(xiàn)服務(wù)掛掉，網(wǎng)絡(luò)閃斷或者超時(shí)，那Boe果此時(shí)RocketMQ將會(huì)不停的詢問(wèn)發(fā)送方來(lái)獲取本地事務(wù)的執(zhí)行狀態(tài)(即事務(wù)回查根據(jù)事務(wù)回查的結(jié)果來(lái)決定Commit或Rollback時(shí)失敗。以上主干流程已由RocetM查的方法即可，具體來(lái)說(shuō)就是實(shí)現(xiàn)下面這個(gè)接口：發(fā)送prepare@parammsg回傳的消息，利用transactionId即可獲取到該消息的唯一@paramarg調(diào)用send方法時(shí)傳遞的參數(shù)，當(dāng)send時(shí)候若有額外的參數(shù)可以傳遞到send方法中，@return返回事務(wù)狀態(tài)，COMMIT：提交ROLLBACK：回滾UNKNOWObjectarg);2.行成功如果是出現(xiàn)了異常，RocetMQ是超時(shí)，2.行成功如果是出現(xiàn)了異常，RocetMQ是超時(shí)，RocetMQ就會(huì)無(wú)限制的消費(fèi)消息，不斷的去執(zhí)行本地事務(wù)，直到成功為止。本實(shí)例通過(guò)RocetMQ的銀行(張三在bank、李四在bank2)，bank1、bank2是兩個(gè)相互獨(dú)立的微服務(wù)。JDK：64位jdk1.8+@parammsg通過(guò)獲取transactionId@return返回事務(wù)狀態(tài)，COMMIT：提交ROLLBACK：回滾UNKNOWLocalTransactionStatecheckLocalTransaction(finalMessageExtRocketMQ客戶端：RocketMQ-spring-boot-starter.2.0.2-本案例需要兩個(gè)數(shù)據(jù)庫(kù)，一個(gè)是bank1，一個(gè)是bank2，無(wú)需創(chuàng)建，直接使用RocketMQ客戶端：RocketMQ-spring-boot-starter.2.0.2-本案例需要兩個(gè)數(shù)據(jù)庫(kù)，一個(gè)是bank1，一個(gè)是bank2，無(wú)需創(chuàng)建，直接使用Hmily快速入門案例中的數(shù)據(jù)庫(kù)即可。另外，為了實(shí)現(xiàn)冪等性，需要分別在bank1、bank2數(shù)據(jù)庫(kù)中新增de_duplication交易記錄表(去重表)。DROPTABLEIFEXISTS`de_duplication`;CREATETABLE`de_duplication`(`tx_no`bigint(20)NOT`create_time`datetime(0)NULLDEFAULTNULL,PRIMARYKEY(`tx_no`)USINGBTREE)ENGINE=InnoDBCHARACTERSET=utf8COLLATE=utf8_binROW_FORMAT=啟動(dòng)啟動(dòng)setROCKETMQ_HOME=[RocketMQ服務(wù)端解壓路徑start[RocketMQ服務(wù)端解壓路徑啟動(dòng)setROCKETMQ_HOME=[RocketMQ服務(wù)端解壓路徑start[RocketMQ服務(wù)端解壓路徑]/bin/mqbroker.cmd-n:9876Maven消息發(fā)送方1.publicclasspublicclassAccountChangeEventimplementsSerializableprivateStringprivatedoubleprivatelong2.publicinterfaceAccountInfoDao@paramaccountNo@paramamount@Update("updateaccount_infosetaccount_balance=account_balance+#{amount}whereaccount_no=#{accountNo}")intupdateAccountBalance(@Param("accountNo")StringaccountNo,@Param("amount")Doubleamount);@paramaccountNo@Select("select*fromaccount_infowherewhereaccount_no=#{accountNo}")AccountInfofindByIdAccountNo(@Param("accountNo")StringaccountNo);@paramtxNo@Select("selectcount(1)fromde_duplicationwheretx_no=#{txNo}")intisExistTx(longtxNo);@paramtxNo3.publicclassBankMessageProducer3.publicclassBankMessageProducer{privateRocketMQTemplatepublicvoidsendAccountChangeEvent(AccountChangeEventaccountChangeEvent)1.JSONObjectobject=newJSONObject();object.put("accountChange",accountChangeEvent);Message<String>msg=//2.發(fā)送消息msg,null);4.實(shí)現(xiàn)業(yè)務(wù)層代碼，分別實(shí)現(xiàn)了發(fā)送事務(wù)消息與本地事務(wù)扣減金額，注意更新帳號(hào)余額-@paramvoidupdateAccountBalance(AccountChangeEvent更新帳號(hào)余額-@paramvoiddoUpdateAccountBalance(AccountChangeEventpublicclassAccountInfoServiceImplimplementsAccountInfoServiceprivateBankMessageProducer更新帳號(hào)余額-@paramintaddTx(long更新帳號(hào)余額-@param@Transactional(isolation=publicvoiddoUpdateAccountBalance(AccountChangeEventaccountChange)getAmount()*-1);5.實(shí)現(xiàn)RocketMQexecuteLocalTransaction，該方法執(zhí)行本地事務(wù)，會(huì)被RocketMQ@RocketMQTransactionListener(txProducerGroup="producer_ensure_transfer")publicclassTransferTransactionListenerImplimplementsRocketMQLocalTransactionListener{privateAccountInfoService@param@parampublicRocketMQLocalTransactionStateexecuteLocalTransaction(Messagemsg,Objectarg){//1.finalJSONObjectjsonObject=JSON.parseObject(newString((byte[])AccountChangeEventaccountChangeEvent//2.BooleanisCommit=true;try{//2.BooleanisCommit=true;try{}catch(Exceptione){isCommit=return}elsereturn@parampublicRocketMQLocalTransactionStatecheckLocalTransaction(Message//1.finalJSONObjectjsonObject=JSON.parseObject(newString((byte[])AccountChangeEventaccountChangeEvent//2.查詢de_duplicationintisExistTx=return}elsereturn6.完善ControllerpublicclassAccountInfoController{privateAccountInfoServicepublicclassAccountInfoController{privateAccountInfoServicepublicStringtransfer(){return"轉(zhuǎn)賬成功實(shí)現(xiàn)數(shù)據(jù)訪問(wèn)層，和bank1實(shí)現(xiàn)業(yè)務(wù)層功能,增加賬號(hào)余額，注意這里使用了交易記錄去重表（de_duplicatio）控制@paramvoidupdateAccountBalance(AccountChangeEventpublicclassAccountInfoServiceImplimplementsAccountInfoService@Transactional(isolation=publicvoidupdateAccountBalance(AccountChangeEventaccountChange){intisExistTx=accountInfoDao.isExistTx(accountChange.getTxNo());if(isExistTx==0){3.實(shí)現(xiàn)RocketMQ事務(wù)消息監(jiān)聽(tīng)器,5.3.4bank1和bank2bank1執(zhí)行本地事務(wù)失敗，則5.3.4bank1和bank2bank1執(zhí)行本地事務(wù)失敗，則bank2為基于消息執(zhí)行的異步操作,避免了分布式事務(wù)中的同步阻塞操作的影響，并實(shí)現(xiàn)了兩個(gè)服務(wù)的解耦。privateAccountInfoServicepublicvoidonMessage(StringprojectStr){System.out.println("開始消費(fèi)消息："+projectStr);AccountChangeEventaccountChangeEvent=在滿標(biāo)放款業(yè)務(wù)中(上圖紅框部分在滿標(biāo)放款業(yè)務(wù)中(上圖紅框部分)，交易中心修改標(biāo)的狀態(tài)為“還款中”劃和應(yīng)收明細(xì)。兩者為原子性綁定，即：標(biāo)的狀態(tài)修改成功，就必須生成還款計(jì)劃和應(yīng)收明細(xì)。由于涉及到兩個(gè)獨(dú)立的微服務(wù)，這里就存在分布式事務(wù)問(wèn)題。業(yè)務(wù)對(duì)強(qiáng)一致性要求較低，因此我們可以采用RocetMQ可靠消息實(shí)現(xiàn)最終一致性這個(gè)解決方案。1.檢查pom.xml1.檢查pom.xml<artifactId>rocketmq-spring-boot-2.檢查Apollo中交易中心項(xiàng)目的micro_ducer.group=3.新建P2pTransactionProducerpublicv