?????????????????????????????
?????????????????????????????

別問，問就是不行

分布式事務(wù)你應(yīng)該是知道的。但是這個多線程事務(wù)......

沒事，我慢慢給你說。

如圖所示，有個小伙伴想要實(shí)現(xiàn)多線程事務(wù)。

這個需求其實(shí)我在不同的地方看到過很多次，所以我才說：這個問題又出現(xiàn)了。

那么有解決方案嗎？

在此之前，我的回答都是非常的肯定：毋庸置疑，做不了，肯定是沒有的。

為什么呢？

我們先從理論上去推理一下。

來，首先我問你，事務(wù)的特性是什么？

這個不難吧？八股文必背內(nèi)容之一，ACID 必須張口就來：

• 原子性（Atomicity）
• 一致性（Consistency）
• 隔離性（Isolation）
• 持久性（Durability）

那么我又問你：你覺得如果真的實(shí)現(xiàn)了多線程事務(wù)，那么我們破壞了事務(wù)的哪個特性？

多線程事務(wù)你也別想的多深奧，你就想，兩個不同的用戶各自發(fā)起了一個下單請求，這個請求對應(yīng)的后臺實(shí)現(xiàn)邏輯中是有事務(wù)存在的。

兩個用戶，兩個線程，這不就是多線程事務(wù)嗎？

這種場景下你沒有想過怎么分別去控制兩個用戶的事務(wù)操作吧？

因?yàn)檫@兩個操作之間就是完全隔離的，各自拿著各自的鏈接玩兒。

所以多個事務(wù)之間的最基本的原則是什么？

隔離性。兩個事務(wù)操作之間不應(yīng)該相互干擾。

而多線程事務(wù)想要實(shí)現(xiàn)的是 A 線程異常了。A，B 線程的事務(wù)一起回滾。

事務(wù)的特性里面就卡的死死的。所以，多線程事務(wù)從理論上就是行不通的。

通過理論指導(dǎo)實(shí)踐，那么多線程事務(wù)的代碼也就是寫不出來的。

前面說到隔離性。那么請問，Spring 的源碼里面，對于事務(wù)的隔離性是如何保證的呢？

答案就是 ThreadLocal。

在事務(wù)開啟的時候，把當(dāng)前的鏈接保存在了 ThreadLocal 里面，從而保證了多線程之間的隔離性：

org.springframework.transaction.support.TransactionSynchronizationManager

可以看到，這個 resource 對象是一個 ThreadLocal 對象。

在下面這個方法中進(jìn)行了賦值操作：

org.springframework.jdbc.datasource.DataSourceTransactionManager#doBegin

其中的 bindResource 方法中，就是把數(shù)據(jù)庫鏈接綁定到當(dāng)前線程中，其中的 resource 就是我們剛剛說的 ThreadLocal：

這樣，用 ThreadLocal? 保證了各個線程各自玩自己的。

我們不可能打破 ThreadLocal 的使用規(guī)則，讓各個線程共享同一個 ThreadLocal 吧？

鐵子，你要是這樣去做的話，硬寫也是寫的出來的，但是總感覺是不是走的有點(diǎn)遠(yuǎn)了？

所以，無論從理論上，還是代碼實(shí)現(xiàn)上，我都認(rèn)為這個需求是不能實(shí)現(xiàn)的。

至少我之前是這樣想的。

但是事情，稍稍的發(fā)生了一點(diǎn)點(diǎn)的變化。

? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ? ?

說個場景，常規(guī)實(shí)現(xiàn)

???????????????????????????????????????????????????????任何脫離場景討論技術(shù)實(shí)現(xiàn)的行為都是耍流氓。

所以，我們先看一下場景是什么。

假設(shè)我們有一個大數(shù)據(jù)系統(tǒng)，每天指定時間，我們就需要從大數(shù)據(jù)系統(tǒng)中拉取 50w 條數(shù)據(jù)，對數(shù)據(jù)進(jìn)行一個清洗操作，然后把數(shù)據(jù)保存到我們業(yè)務(wù)系統(tǒng)的數(shù)據(jù)庫中。

對于業(yè)務(wù)系統(tǒng)而言，這 50w 條數(shù)據(jù)，必須全部落庫，差一條都不行。要么就是一條都不插入。

在這個過程中，不會去調(diào)用其他的外部接口，也不會有其他的流程去操作這個表的數(shù)據(jù)。

既然說到一條不差了，那么對于大家直觀而言，想到的肯定是兩個解決方案：

1. 開啟事務(wù)，然后在 for 循環(huán)中慢慢的插入。
2. 直接一條語句批量插入。

我們先說第一個。

對于這種需求，開啟事務(wù)，然后在 for 循環(huán)中一條條的插入可以說是非常 low 的解決方案了。

效率非常的低下，給大家演示一下。

比如，我們有一個 Student 表，表結(jié)構(gòu)非常簡單，如下：

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)目中，我們通過 for 循環(huán)插入數(shù)據(jù)，同時該方法上有 @Transactional 注解：

num 參數(shù)是我們通過前端請求傳遞過來的數(shù)據(jù)，代表要插入 num 條數(shù)據(jù)：

這種情況下，我們可以通過下面的鏈接，模擬插入指定數(shù)量的數(shù)據(jù)：

http://127.0.0.1:8081/insertOneByOne?num=xxx

我嘗試了把 num 設(shè)置為 50w，讓它慢慢的跑著，但是我還是太年輕了，等了非常長的時間都沒有等到結(jié)果。

于是我把 num 改為了 5000，運(yùn)行結(jié)果如下：

insertOneByOne執(zhí)行耗時:133449ms,num=5000

一條條的插入 5000 條數(shù)據(jù)，耗時 133.5 s 的樣子。

按照這個速度，插入 50w 條數(shù)據(jù)得 13350s，大概也是這么多小時：

這誰頂?shù)米“　?/p>

所以，這方案擁有巨大的優(yōu)化空間。

比如我們優(yōu)化為這樣的批量插入：

其對應(yīng)的 sql 語句是這樣的：

insert into table ([列名],[列名]) VALUES ([列值],[列值]), ([列值],[列值]);

我們還是通過前端接口調(diào)用：

當(dāng)我們的 num 設(shè)置為 5000 的時候，我頁面刷新了 10 次，你看批量插入的耗時基本上在 200ms 毫秒以內(nèi)：

從 133.5s 到 200ms，朋友們，這是什么東西？

這是降維的打擊，質(zhì)上的飛躍啊。性能提升了近 667 倍的樣子。

為什么批量插入能有這么大的飛躍呢？

你想啊，之前 for 循環(huán)插入，一個連接，一次事務(wù)。

雖然 SpringBoot 2.0 默認(rèn)使用了 HikariPool，連接池里面默認(rèn)給你搞 10 個連接，我只用一個。

但是也架不住你 5000 次頻繁IO呀。

所以，耗時長是必然的。

而批量插入只是一條 sql 語句，所以只需要一個連接，還不需要開啟事務(wù)。

為啥不用開啟事務(wù)？

你一條 sql 開啟事務(wù)有錘子用啊？

那么，如果我們一口氣插入 50w 條數(shù)據(jù)，會是怎么樣的呢？

來，搞一波，試一下：

http://127.0.0.1:8081/insertBatch?num=500000

哦豁。

可以看到拋出了一個異常。而且錯誤信息非常的清晰：

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.

說你這個包太大了?？梢酝ㄟ^設(shè)置 max_allowed_packet 來改變包大小。

我們可以通過下面的語句查詢當(dāng)前的配置大?。?/p>

select @@max_allowed_packet;

可以看到是 1048576 字節(jié)，即 1024*1024，1M 大小。

而我們需要傳輸?shù)陌笮∈?42777840 字節(jié)，大概是 41M 的樣子。

所以我們需要修改配置大小。

這個地方也給大家提了個醒：如果你的 sql 語句非常大，里面有大字段，記得調(diào)整一下 mysql 的這個參數(shù)。

可以通過修改配置文件或者直接執(zhí)行 sql 語句的方式進(jìn)行修改。

我這里就使用 sql 語句把值修改為 64M：

set global max_allowed_packet = 1024*1024*64;

然后再次執(zhí)行，可以看到插入成功了：

50w 的數(shù)據(jù)，74s 的樣子。

數(shù)據(jù)要么全部提交，要么一條也沒有，需求也實(shí)現(xiàn)了。

時間上呢，是有點(diǎn)長，但是好像也想不到什么好的提升方案。

那么我們怎么還能再縮短點(diǎn)時間呢？

騷想法出現(xiàn)了

要是想讓時間再短一點(diǎn)，我能想到的一個解決方案，也就是祭出多線程了。

50w 數(shù)據(jù)。我們開五個線程，一個線程處理 10w 數(shù)據(jù)，沒有異常就保存入庫，出現(xiàn)問題就回滾。

這個需求很好實(shí)現(xiàn)。分分鐘就能寫出來。

但是再加上一個需求：這 5 個線程的數(shù)據(jù)，如果有一個線程出現(xiàn)問題了，需要全部回滾。

順著思路慢慢擼，我們發(fā)現(xiàn)這個時候就是所謂的多線程事務(wù)了。

我之前說完全不可能實(shí)現(xiàn)，是因?yàn)樘岬绞聞?wù)我就想到了 @Transactional 注解去實(shí)現(xiàn)了。

我們只需要正確使用它，然后關(guān)心業(yè)務(wù)邏輯即可。

只要你的程序沒有 bug ，你就不需要也根本插手不了事務(wù)的開啟和提交或者回滾。

這種代碼的寫法我們叫做聲明式事務(wù)。

和聲明式事務(wù)對應(yīng)的就是編程式事務(wù)了。

通過編程式事務(wù)，我們就能完全掌控事務(wù)的開啟和提交或者回滾操作。

能想到編程式事務(wù)，這事基本上就成了一半了。

我先給你說一下思路。

你想，首先假設(shè)我們有一個全局變量為 Boolean 類型，默認(rèn)為true，含義為可以提交事務(wù)。

然后我們開啟 5 個子線程，各自處理 10w 條數(shù)據(jù)。

在子線程里面，我們可以先通過編程式事務(wù)開啟事務(wù)，插入 10w 條數(shù)據(jù)后不進(jìn)行提交。同時告訴主線程，我這邊準(zhǔn)備好了，進(jìn)入等待。

如果子線程里面出現(xiàn)了異常，那么我就告訴主線程，我這邊出問題了，然后自己進(jìn)行回滾。

不論怎樣，主線程都會收集到 5 個子線程的狀態(tài)。主線程檢測到，如果有一個線程出現(xiàn)了問題，那么設(shè)置全局變量為 false，含義為回滾事務(wù)。

然后喚醒所有等待的子線程，進(jìn)行回滾。

根據(jù)上面的流程，寫出模擬代碼就是這樣的，大家可以直接復(fù)制出來運(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()?+?"：開始執(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)異常,開始事務(wù)回滾");
????????????????}
????????????});
????????}
????????//主線程等待所有子線程執(zhí)行response
????????try?{
????????????childMonitor.await();
????????????for?(Boolean?resp?:?childResponse)?{
????????????????if?(!resp)?{
????????????????????//如果有一個子線程執(zhí)行失敗了，則改變mainResult，讓所有子線程回滾
????????????????????System.out.println(Thread.currentThread().getName()+":有線程執(zhí)行失敗，標(biāo)志位設(shè)置為false");
????????????????????IS_OK?=?false;
????????????????????break;
????????????????}
????????????}
????????????//主線程獲取結(jié)果成功，讓子線程開始根據(jù)主線程的結(jié)果執(zhí)行（提交或回滾）
????????????mainMonitor.countDown();
????????????//為了讓主線程阻塞，讓子線程執(zhí)行。
????????????Thread.currentThread().join();
????????}?catch?(Exception?e)?{
????????????e.printStackTrace();
????????}
????}
}

在所有子線程都正常的情況下，輸出結(jié)果是這樣的：

從結(jié)果看，是符合我們的預(yù)期的。

假設(shè)有子線程出現(xiàn)了異常，那么運(yùn)行結(jié)果是這樣的：

一個線程出現(xiàn)異常，全部線程都進(jìn)行回滾，這樣看來也是符合預(yù)期的。

如果你根據(jù)前面的需求寫出了這樣的代碼，那么恭喜你，一不留神實(shí)現(xiàn)了一個類似于兩階段提交（2PC）的一致性協(xié)議。

我前面說的能想到編程式事務(wù)，這事基本上就成了一半了。

而另外一半，就是兩階段提交（2PC）。

依瓢畫葫蘆

有了前面的瓢，你照著畫個葫蘆不是很簡單的事情嗎？

就不大段上代碼了，示例代碼可以點(diǎn)擊閱讀原文獲取到，所以我這里截個圖吧：

上面的代碼應(yīng)該是非常好理解的，開啟五個線程，每個線程插入 10w 條數(shù)據(jù)。

這個不用說，用腳趾頭想也能知道，肯定是比一次性批量插入 50w 條數(shù)據(jù)快的。

至于快多少，不廢話了，直接看執(zhí)行效果吧。

由于我們的 controller 是這樣的：

所以調(diào)用鏈接：

http://127.0.0.1:8081/batchHandle

輸出結(jié)果如下：

還記得我們批量插入的耗時嗎？

73791ms。

從 73791ms 到 15719ms。快了 58s 的樣子。

已經(jīng)非常不錯了。

那么如果是某個線程拋出了異常呢？比如這樣：

我們看看日志輸出：

通過日志分析，看起來也是符合要求的。

而從讀者反饋的實(shí)際測試效果來看，也是非常顯著的：

真的符合要求嗎？

符合要求，只是看起來而已。

經(jīng)驗(yàn)老道的讀者朋友們肯定早就看到問題所在了。已經(jīng)把手舉得高高的：老師，這題我知道。

我之前說了，這個實(shí)現(xiàn)方式實(shí)際上就是編程式事務(wù)配合二階段提交（2PC）使用。

破綻就出在 2PC 上。

就像我和讀者討論這樣的：

所以，就算在你代碼寫的沒有任何 BUG 的前提下，還是保證不了數(shù)據(jù)一致性。

到這不能再往后扯了，再往后就是 3PC，TTC，Seata... 這一套分布式事務(wù)的東西了。

這套東西寫下來，就得上萬字了。

所以我從海神那邊轉(zhuǎn)了一篇文章，放在第二條推送里面了。如果大家有興趣的可以去看一下。干貨滿滿。

其實(shí)當(dāng)我們把上面的一個個子線程理解為微服務(wù)中的一個個子系統(tǒng)的時候，這就是一個分布式事務(wù)的場景了。

而我們拿出來的解決方案，并不是一個完美的解決方案。

雖然，從某種角度上，我們繞開了事務(wù)的隔離性，但是有一定概率出現(xiàn)數(shù)據(jù)一致性問題，即使概率比較小。

所以我稱這種方案為：基于運(yùn)氣編程，用運(yùn)氣換時間。

注意事項(xiàng)

?關(guān)于上面的代碼，其實(shí)還有幾個需要注意的地方。

給大家提個醒。

第一個：啟用多少線程進(jìn)行分配數(shù)據(jù)插入，這個參數(shù)是可以進(jìn)行調(diào)整的。

比如我修改為 10 個線程，每個線程插入 5w 條數(shù)據(jù)。那么執(zhí)行時間又快了 2s：

但是一定記得不是越大越好，同時記得調(diào)整數(shù)據(jù)庫連接池的最大連接數(shù)。

不然白搭。

第二個：正是因?yàn)閱佣嗌倬€程是可以進(jìn)行調(diào)整的，甚至是可以每次進(jìn)行計算的。

那么必須要注意的一個問題是不能讓任何一個任務(wù)進(jìn)入隊(duì)列里面。一旦進(jìn)入隊(duì)列，程序立馬就涼。

你想，如果我們需要開啟 5 個子線程，但是核心線程數(shù)只有 4 個，有一個任務(wù)進(jìn)入隊(duì)列了。

那么這 4 個核心線程會一直阻塞住，等待主線程喚醒。

而主線程這個時候在干什么？

在等 5 個線程的運(yùn)行結(jié)果，但是它只能收集到 4 個結(jié)果。

所以它會一直等下去。

第三個：這里是多個線程開啟了事務(wù)在往表里插入數(shù)據(jù)，謹(jǐn)防數(shù)據(jù)庫死鎖。

第四個：注意程序里面的代碼，countDown 安裝標(biāo)準(zhǔn)寫法上是要放到 finally 代碼塊里面的，我這里為了截圖的美觀度，省去了這個步驟：

你如果真的要用，得注意一下。而且這個finally你得想清楚了寫，不是隨便寫的。

第五個：我這里只是提供一個思路，而且它也根本不是什么多線程事務(wù)。

也再次證明了，多線程事務(wù)就是一個偽命題。

所以我給出一個基于運(yùn)氣的偽一致性的回答也不過分吧。

第六個：多線程事務(wù)換個角度想，可以理解為分布式事務(wù)。

那么可以借助這個案例去了解分布式事務(wù)。

但是解決分布式事務(wù)的最好的方法就是：不要有分布式事務(wù)！

而解決分布式事務(wù)的絕大部分落地方案都是：最終一致性。

性價比高，大多數(shù)業(yè)務(wù)上也能接受。

第七個：這個解決方案你要拿到生產(chǎn)用的話，記得先和業(yè)務(wù)同事溝通好，能不能接受這種情況。速度和安全之間的兩難抉擇。

同時自己留好人工修數(shù)的接口：