要我說(shuō)，多線程事務(wù)它必須就是個(gè)偽命題！
點(diǎn)擊上方?好好學(xué)java?，選擇?星標(biāo)?公眾號(hào)
重磅資訊，干貨，第一時(shí)間送達(dá) 今日推薦：這款I(lǐng)DEA插件刷爆了朋友圈，網(wǎng)友：這用起來(lái)有點(diǎn)酸爽～
個(gè)人原創(chuàng)100W +訪問(wèn)量博客：點(diǎn)擊前往，查看更多
別問(wèn)，問(wèn)就是不行
分布式事務(wù)你應(yīng)該是知道的。但是這個(gè)多線程事務(wù)......
沒(méi)事，我慢慢給你說(shuō)。
如圖所示，有個(gè)小伙伴想要實(shí)現(xiàn)多線程事務(wù)。
這個(gè)需求其實(shí)我在不同的地方看到過(guò)很多次，所以我才說(shuō)：這個(gè)問(wèn)題又出現(xiàn)了。
那么有解決方案嗎？
在此之前，我的回答都是非常的肯定：毋庸置疑，做不了，肯定是沒(méi)有的。
為什么呢？
我們先從理論上去推理一下。
來(lái)，首先我問(wèn)你，事務(wù)的特性是什么？
這個(gè)不難吧？八股文必背內(nèi)容之一，ACID 必須張口就來(lái)：
原子性（Atomicity） 一致性（Consistency） 隔離性（Isolation） 持久性（Durability）
那么我又問(wèn)你：你覺(jué)得如果真的實(shí)現(xiàn)了多線程事務(wù)，那么我們破壞了事務(wù)的哪個(gè)特性？
多線程事務(wù)你也別想的多深?yuàn)W，你就想，兩個(gè)不同的用戶各自發(fā)起了一個(gè)下單請(qǐng)求，這個(gè)請(qǐng)求對(duì)應(yīng)的后臺(tái)實(shí)現(xiàn)邏輯中是有事務(wù)存在的。
兩個(gè)用戶，兩個(gè)線程，這不就是多線程事務(wù)嗎？
這種場(chǎng)景下你沒(méi)有想過(guò)怎么分別去控制兩個(gè)用戶的事務(wù)操作吧？
因?yàn)檫@兩個(gè)操作之間就是完全隔離的，各自拿著各自的鏈接玩兒。
所以多個(gè)事務(wù)之間的最基本的原則是什么？
隔離性。兩個(gè)事務(wù)操作之間不應(yīng)該相互干擾。
而多線程事務(wù)想要實(shí)現(xiàn)的是 A 線程異常了。A，B 線程的事務(wù)一起回滾。
事務(wù)的特性里面就卡的死死的。所以，多線程事務(wù)從理論上就是行不通的。
通過(guò)理論指導(dǎo)實(shí)踐，那么多線程事務(wù)的代碼也就是寫(xiě)不出來(lái)的。
前面說(shuō)到隔離性。那么請(qǐng)問(wèn)，Spring 的源碼里面，對(duì)于事務(wù)的隔離性是如何保證的呢？
答案就是 ThreadLocal。
在事務(wù)開(kāi)啟的時(shí)候，把當(dāng)前的鏈接保存在了 ThreadLocal 里面，從而保證了多線程之間的隔離性：
org.springframework.transaction.support.TransactionSynchronizationManager
可以看到，這個(gè) resource 對(duì)象是一個(gè) ThreadLocal 對(duì)象。
在下面這個(gè)方法中進(jìn)行了賦值操作：
org.springframework.jdbc.datasource.DataSourceTransactionManager#doBegin
其中的 bindResource 方法中，就是把數(shù)據(jù)庫(kù)鏈接綁定到當(dāng)前線程中，其中的 resource 就是我們剛剛說(shuō)的 ThreadLocal：
這樣，用 ThreadLocal? 保證了各個(gè)線程各自玩自己的。
我們不可能打破 ThreadLocal 的使用規(guī)則，讓各個(gè)線程共享同一個(gè) ThreadLocal 吧？
鐵子，你要是這樣去做的話，硬寫(xiě)也是寫(xiě)的出來(lái)的，但是總感覺(jué)是不是走的有點(diǎn)遠(yuǎn)了？
所以，無(wú)論從理論上，還是代碼實(shí)現(xiàn)上，我都認(rèn)為這個(gè)需求是不能實(shí)現(xiàn)的。
至少我之前是這樣想的。
但是事情，稍稍的發(fā)生了一點(diǎn)點(diǎn)的變化。
說(shuō)個(gè)場(chǎng)景，常規(guī)實(shí)現(xiàn)
任何脫離場(chǎng)景討論技術(shù)實(shí)現(xiàn)的行為都是耍流氓。
所以，我們先看一下場(chǎng)景是什么。
假設(shè)我們有一個(gè)大數(shù)據(jù)系統(tǒng)，每天指定時(shí)間，我們就需要從大數(shù)據(jù)系統(tǒng)中拉取 50w 條數(shù)據(jù)，對(duì)數(shù)據(jù)進(jìn)行一個(gè)清洗操作，然后把數(shù)據(jù)保存到我們業(yè)務(wù)系統(tǒng)的數(shù)據(jù)庫(kù)中。
對(duì)于業(yè)務(wù)系統(tǒng)而言，這 50w 條數(shù)據(jù)，必須全部落庫(kù)，差一條都不行。要么就是一條都不插入。
在這個(gè)過(guò)程中，不會(huì)去調(diào)用其他的外部接口，也不會(huì)有其他的流程去操作這個(gè)表的數(shù)據(jù)。
既然說(shuō)到一條不差了，那么對(duì)于大家直觀而言，想到的肯定是兩個(gè)解決方案：
開(kāi)啟事務(wù)，然后在 for 循環(huán)中慢慢的插入。 直接一條語(yǔ)句批量插入。
我們先說(shuō)第一個(gè)。
對(duì)于這種需求，開(kāi)啟事務(wù)，然后在 for 循環(huán)中一條條的插入可以說(shuō)是非常 low 的解決方案了。
效率非常的低下，給大家演示一下。
比如，我們有一個(gè) Student 表，表結(jié)構(gòu)非常簡(jiǎn)單，如下：
CREATE?TABLE?`student`?(
??`id`?bigint(63)?NOT?NULL?AUTO_INCREMENT,
??`name`?varchar(32)?DEFAULT?NULL,
??`home`?varchar(64)?DEFAULT?NULL,
??PRIMARY?KEY?(`id`)
)?ENGINE=InnoDB?AUTO_INCREMENT=1?DEFAULT?CHARSET=utf8;
在我們的項(xiàng)目中，我們通過(guò) for 循環(huán)插入數(shù)據(jù)，同時(shí)該方法上有 @Transactional 注解：
num 參數(shù)是我們通過(guò)前端請(qǐng)求傳遞過(guò)來(lái)的數(shù)據(jù)，代表要插入 num 條數(shù)據(jù)：
這種情況下，我們可以通過(guò)下面的鏈接，模擬插入指定數(shù)量的數(shù)據(jù)：
http://127.0.0.1:8081/insertOneByOne?num=xxx
我嘗試了把 num 設(shè)置為 50w，讓它慢慢的跑著，但是我還是太年輕了，等了非常長(zhǎng)的時(shí)間都沒(méi)有等到結(jié)果。
于是我把 num 改為了 5000，運(yùn)行結(jié)果如下：
insertOneByOne執(zhí)行耗時(shí):133449ms,num=5000
一條條的插入 5000 條數(shù)據(jù)，耗時(shí) 133.5 s 的樣子。
按照這個(gè)速度，插入 50w 條數(shù)據(jù)得 13350s，大概也是這么多小時(shí)：
這誰(shuí)頂?shù)米“ ?/p>
所以，這方案擁有巨大的優(yōu)化空間。
比如我們優(yōu)化為這樣的批量插入：
其對(duì)應(yīng)的 sql 語(yǔ)句是這樣的：
insert into table ([列名],[列名]) VALUES ([列值],[列值]), ([列值],[列值]);
我們還是通過(guò)前端接口調(diào)用：
當(dāng)我們的 num 設(shè)置為 5000 的時(shí)候，我頁(yè)面刷新了 10 次，你看批量插入的耗時(shí)基本上在 200ms 毫秒以內(nèi)：
從 133.5s 到 200ms，朋友們，這是什么東西？
這是降維的打擊，質(zhì)上的飛躍啊。性能提升了近 667 倍的樣子。
為什么批量插入能有這么大的飛躍呢？
你想啊，之前 for 循環(huán)插入，一個(gè)連接，一次事務(wù)。
雖然 SpringBoot 2.0 默認(rèn)使用了 HikariPool，連接池里面默認(rèn)給你搞 10 個(gè)連接，我只用一個(gè)。
但是也架不住你 5000 次頻繁IO呀。
所以，耗時(shí)長(zhǎng)是必然的。
而批量插入只是一條 sql 語(yǔ)句，所以只需要一個(gè)連接，還不需要開(kāi)啟事務(wù)。
為啥不用開(kāi)啟事務(wù)？
你一條 sql 開(kāi)啟事務(wù)有錘子用啊？
那么，如果我們一口氣插入 50w 條數(shù)據(jù)，會(huì)是怎么樣的呢？
來(lái)，搞一波，試一下：
http://127.0.0.1:8081/insertBatch?num=500000
哦豁。
可以看到拋出了一個(gè)異常。而且錯(cuò)誤信息非常的清晰：
Packet?for?query?is?too?large?(42777840?>?1048576).?You?can?change?this?value?on?the?server?by?setting?the?max_allowed_packet'?variable.;?nested?exception?is?com.mysql.jdbc.PacketTooBigException:?Packet?for?query?is?too?large?(42777840?>?1048576).You?can?change?this?value?on?the?server?by?setting?the?max_allowed_packet'?variable.說(shuō)你這個(gè)包太大了。可以通過(guò)設(shè)置 max_allowed_packet 來(lái)改變包大小。
我們可以通過(guò)下面的語(yǔ)句查詢當(dāng)前的配置大小：
select @@max_allowed_packet;
可以看到是 1048576 字節(jié)，即 1024*1024，1M 大小。
而我們需要傳輸?shù)陌笮∈?42777840 字節(jié)，大概是 41M 的樣子。
所以我們需要修改配置大小。
這個(gè)地方也給大家提了個(gè)醒：如果你的 sql 語(yǔ)句非常大，里面有大字段，記得調(diào)整一下 mysql 的這個(gè)參數(shù)。
可以通過(guò)修改配置文件或者直接執(zhí)行 sql 語(yǔ)句的方式進(jìn)行修改。
我這里就使用 sql 語(yǔ)句把值修改為 64M：
set global max_allowed_packet = 1024*1024*64;
然后再次執(zhí)行，可以看到插入成功了：
50w 的數(shù)據(jù)，74s 的樣子。
數(shù)據(jù)要么全部提交，要么一條也沒(méi)有，需求也實(shí)現(xiàn)了。
時(shí)間上呢，是有點(diǎn)長(zhǎng)，但是好像也想不到什么好的提升方案。
那么我們?cè)趺催€能再縮短點(diǎn)時(shí)間呢？
騷想法出現(xiàn)了
要是想讓時(shí)間再短一點(diǎn)，我能想到的一個(gè)解決方案，也就是祭出多線程了。
50w 數(shù)據(jù)。我們開(kāi)五個(gè)線程，一個(gè)線程處理 10w 數(shù)據(jù)，沒(méi)有異常就保存入庫(kù)，出現(xiàn)問(wèn)題就回滾。
這個(gè)需求很好實(shí)現(xiàn)。分分鐘就能寫(xiě)出來(lái)。
但是再加上一個(gè)需求：這 5 個(gè)線程的數(shù)據(jù)，如果有一個(gè)線程出現(xiàn)問(wèn)題了，需要全部回滾。
順著思路慢慢擼，我們發(fā)現(xiàn)這個(gè)時(shí)候就是所謂的多線程事務(wù)了。
我之前說(shuō)完全不可能實(shí)現(xiàn)，是因?yàn)樘岬绞聞?wù)我就想到了 @Transactional 注解去實(shí)現(xiàn)了。
我們只需要正確使用它，然后關(guān)心業(yè)務(wù)邏輯即可。
只要你的程序沒(méi)有 bug ，你就不需要也根本插手不了事務(wù)的開(kāi)啟和提交或者回滾。
這種代碼的寫(xiě)法我們叫做聲明式事務(wù)。
和聲明式事務(wù)對(duì)應(yīng)的就是編程式事務(wù)了。
通過(guò)編程式事務(wù)，我們就能完全掌控事務(wù)的開(kāi)啟和提交或者回滾操作。
能想到編程式事務(wù)，這事基本上就成了一半了。
我先給你說(shuō)一下思路。
你想，首先假設(shè)我們有一個(gè)全局變量為 Boolean 類型，默認(rèn)為true，含義為可以提交事務(wù)。
然后我們開(kāi)啟 5 個(gè)子線程，各自處理 10w 條數(shù)據(jù)。
在子線程里面，我們可以先通過(guò)編程式事務(wù)開(kāi)啟事務(wù)，插入 10w 條數(shù)據(jù)后不進(jìn)行提交。同時(shí)告訴主線程，我這邊準(zhǔn)備好了，進(jìn)入等待。
如果子線程里面出現(xiàn)了異常，那么我就告訴主線程，我這邊出問(wèn)題了，然后自己進(jìn)行回滾。
不論怎樣，主線程都會(huì)收集到 5 個(gè)子線程的狀態(tài)。主線程檢測(cè)到，如果有一個(gè)線程出現(xiàn)了問(wèn)題，那么設(shè)置全局變量為 false，含義為回滾事務(wù)。
然后喚醒所有等待的子線程，進(jìn)行回滾。
根據(jù)上面的流程，寫(xiě)出模擬代碼就是這樣的，大家可以直接復(fù)制出來(lái)運(yùn)行：
public?class?MainTest?{
????//是否可以提交
????public?static?volatile?boolean?IS_OK?=?true;
????public?static?void?main(String[]?args)?{
????????//子線程等待主線程通知
????????CountDownLatch?mainMonitor?=?new?CountDownLatch(1);
????????int?threadCount?=?5;
????????CountDownLatch?childMonitor?=?new?CountDownLatch(threadCount);
????????//子線程運(yùn)行結(jié)果
????????List?childResponse?=?new?ArrayList();
????????ExecutorService?executor?=?Executors.newCachedThreadPool();
????????for?(int?i?=?0;?i?????????????int?finalI?=?i;
????????????executor.execute(()?->?{
????????????????try?{
????????????????????System.out.println(Thread.currentThread().getName()?+?"：開(kāi)始執(zhí)行");
//?if?(finalI?==?4)?{
//?throw?new?Exception("出現(xiàn)異常");
//?}
????????????????????TimeUnit.MILLISECONDS.sleep(ThreadLocalRandom.current().nextInt(1000));
????????????????????childResponse.add(Boolean.TRUE);
????????????????????childMonitor.countDown();
????????????????????System.out.println(Thread.currentThread().getName()?+?"：準(zhǔn)備就緒,等待其他線程結(jié)果,判斷是否事務(wù)提交");
????????????????????mainMonitor.await();
????????????????????if?(IS_OK)?{
????????????????????????System.out.println(Thread.currentThread().getName()?+?"：事務(wù)提交");
????????????????????}?else?{
????????????????????????System.out.println(Thread.currentThread().getName()?+?"：事務(wù)回滾");
????????????????????}
????????????????}?catch?(Exception?e)?{
????????????????????childResponse.add(Boolean.FALSE);
????????????????????childMonitor.countDown();
????????????????????System.out.println(Thread.currentThread().getName()?+?"：出現(xiàn)異常,開(kāi)始事務(wù)回滾");
????????????????}
????????????});
????????}
????????//主線程等待所有子線程執(zhí)行response
????????try?{
????????????childMonitor.await();
????????????for?(Boolean?resp?:?childResponse)?{
????????????????if?(!resp)?{
????????????????????//如果有一個(gè)子線程執(zhí)行失敗了，則改變mainResult，讓所有子線程回滾
????????????????????System.out.println(Thread.currentThread().getName()+":有線程執(zhí)行失敗，標(biāo)志位設(shè)置為false");
????????????????????IS_OK?=?false;
????????????????????break;
????????????????}
????????????}
????????????//主線程獲取結(jié)果成功，讓子線程開(kāi)始根據(jù)主線程的結(jié)果執(zhí)行（提交或回滾）
????????????mainMonitor.countDown();
????????????//為了讓主線程阻塞，讓子線程執(zhí)行。
????????????Thread.currentThread().join();
????????}?catch?(Exception?e)?{
????????????e.printStackTrace();
????????}
????}
}
在所有子線程都正常的情況下，輸出結(jié)果是這樣的：
從結(jié)果看，是符合我們的預(yù)期的。
假設(shè)有子線程出現(xiàn)了異常，那么運(yùn)行結(jié)果是這樣的：
一個(gè)線程出現(xiàn)異常，全部線程都進(jìn)行回滾，這樣看來(lái)也是符合預(yù)期的。
如果你根據(jù)前面的需求寫(xiě)出了這樣的代碼，那么恭喜你，一不留神實(shí)現(xiàn)了一個(gè)類似于兩階段提交（2PC）的一致性協(xié)議。
我前面說(shuō)的能想到編程式事務(wù)，這事基本上就成了一半了。
而另外一半，就是兩階段提交（2PC）。
依瓢畫(huà)葫蘆
有了前面的瓢，你照著畫(huà)個(gè)葫蘆不是很簡(jiǎn)單的事情嗎？
就不大段上代碼了，示例代碼可以點(diǎn)擊閱讀原文獲取到，所以我這里截個(gè)圖吧：
上面的代碼應(yīng)該是非常好理解的，開(kāi)啟五個(gè)線程，每個(gè)線程插入 10w 條數(shù)據(jù)。
這個(gè)不用說(shuō)，用腳趾頭想也能知道，肯定是比一次性批量插入 50w 條數(shù)據(jù)快的。
至于快多少，不廢話了，直接看執(zhí)行效果吧。
由于我們的 controller 是這樣的：
所以調(diào)用鏈接：
http://127.0.0.1:8081/batchHandle
輸出結(jié)果如下：
還記得我們批量插入的耗時(shí)嗎？
73791ms。
從 73791ms 到 15719ms。快了 58s 的樣子。
已經(jīng)非常不錯(cuò)了。
那么如果是某個(gè)線程拋出了異常呢？比如這樣：
我們看看日志輸出：
通過(guò)日志分析，看起來(lái)也是符合要求的。
而從讀者反饋的實(shí)際測(cè)試效果來(lái)看，也是非常顯著的：
真的符合要求嗎？
符合要求，只是看起來(lái)而已。
經(jīng)驗(yàn)老道的讀者朋友們肯定早就看到問(wèn)題所在了。已經(jīng)把手舉得高高的：老師，這題我知道。
我之前說(shuō)了，這個(gè)實(shí)現(xiàn)方式實(shí)際上就是編程式事務(wù)配合二階段提交（2PC）使用。
破綻就出在 2PC 上。
就像我和讀者討論這樣的：
所以，就算在你代碼寫(xiě)的沒(méi)有任何 BUG 的前提下，還是保證不了數(shù)據(jù)一致性。
到這不能再往后扯了，再往后就是 3PC，TTC，Seata... 這一套分布式事務(wù)的東西了。
這套東西寫(xiě)下來(lái)，就得上萬(wàn)字了。
所以我從海神那邊轉(zhuǎn)了一篇文章，放在第二條推送里面了。如果大家有興趣的可以去看一下。干貨滿滿。
其實(shí)當(dāng)我們把上面的一個(gè)個(gè)子線程理解為微服務(wù)中的一個(gè)個(gè)子系統(tǒng)的時(shí)候，這就是一個(gè)分布式事務(wù)的場(chǎng)景了。
而我們拿出來(lái)的解決方案，并不是一個(gè)完美的解決方案。
雖然，從某種角度上，我們繞開(kāi)了事務(wù)的隔離性，但是有一定概率出現(xiàn)數(shù)據(jù)一致性問(wèn)題，即使概率比較小。
所以我稱這種方案為：基于運(yùn)氣編程，用運(yùn)氣換時(shí)間。
注意事項(xiàng)
關(guān)于上面的代碼，其實(shí)還有幾個(gè)需要注意的地方。
給大家提個(gè)醒。
第一個(gè)：?jiǎn)⒂枚嗌倬€程進(jìn)行分配數(shù)據(jù)插入，這個(gè)參數(shù)是可以進(jìn)行調(diào)整的。
比如我修改為 10 個(gè)線程，每個(gè)線程插入 5w 條數(shù)據(jù)。那么執(zhí)行時(shí)間又快了 2s：
但是一定記得不是越大越好，同時(shí)記得調(diào)整數(shù)據(jù)庫(kù)連接池的最大連接數(shù)。
不然白搭。
第二個(gè)：正是因?yàn)閱?dòng)多少線程是可以進(jìn)行調(diào)整的，甚至是可以每次進(jìn)行計(jì)算的。
那么必須要注意的一個(gè)問(wèn)題是不能讓任何一個(gè)任務(wù)進(jìn)入隊(duì)列里面。一旦進(jìn)入隊(duì)列，程序立馬就涼。
你想，如果我們需要開(kāi)啟 5 個(gè)子線程，但是核心線程數(shù)只有 4 個(gè)，有一個(gè)任務(wù)進(jìn)入隊(duì)列了。
那么這 4 個(gè)核心線程會(huì)一直阻塞住，等待主線程喚醒。
而主線程這個(gè)時(shí)候在干什么？
在等 5 個(gè)線程的運(yùn)行結(jié)果，但是它只能收集到 4 個(gè)結(jié)果。
所以它會(huì)一直等下去。
第三個(gè)：這里是多個(gè)線程開(kāi)啟了事務(wù)在往表里插入數(shù)據(jù)，謹(jǐn)防數(shù)據(jù)庫(kù)死鎖。
第四個(gè)：注意程序里面的代碼，countDown 安裝標(biāo)準(zhǔn)寫(xiě)法上是要放到 finally 代碼塊里面的，我這里為了截圖的美觀度，省去了這個(gè)步驟：
你如果真的要用，得注意一下。而且這個(gè)finally你得想清楚了寫(xiě)，不是隨便寫(xiě)的。
第五個(gè)：我這里只是提供一個(gè)思路，而且它也根本不是什么多線程事務(wù)。
也再次證明了，多線程事務(wù)就是一個(gè)偽命題。
所以我給出一個(gè)基于運(yùn)氣的偽一致性的回答也不過(guò)分吧。
第六個(gè)：多線程事務(wù)換個(gè)角度想，可以理解為分布式事務(wù)。
那么可以借助這個(gè)案例去了解分布式事務(wù)。
但是解決分布式事務(wù)的最好的方法就是：不要有分布式事務(wù)！
而解決分布式事務(wù)的絕大部分落地方案都是：最終一致性。
性價(jià)比高，大多數(shù)業(yè)務(wù)上也能接受。
第七個(gè)：這個(gè)解決方案你要拿到生產(chǎn)用的話，記得先和業(yè)務(wù)同事溝通好，能不能接受這種情況。速度和安全之間的兩難抉擇。
同時(shí)自己留好人工修數(shù)的接口：
