總結(jié)：Java屆很難得有讀百十行代碼就能增加修煉的機(jī)會(huì)，這里有一個(gè)。

通常，我在看書的時(shí)候一般不寫代碼，因?yàn)槲业哪X袋被設(shè)定成單線程的，一旦同時(shí)喂給它不同的信息，它就無(wú)法處理。

但多線程對(duì)電腦來(lái)說(shuō)就是小菜一碟，它可以同時(shí)做很多事，看起來(lái)匪夷所思。好希望把自己的大腦皮層移植到這些牛x的設(shè)備上。

用人腦思考電腦正在思考的問(wèn)題，這本身就是一種折磨。但平常的工作和面試中，又不得不面對(duì)這樣的場(chǎng)景，所以多線程就成了編程路上一塊難啃的骨頭。

HikariCP是SpringBoot默認(rèn)的數(shù)據(jù)庫(kù)連接池，它毫不謙虛的的起了一個(gè)叫做光的名字，這讓國(guó)產(chǎn)Druid很沒(méi)面子。

還是言歸正傳，看一下Hikari中的ConcurrentBag吧。

核心數(shù)據(jù)結(jié)構(gòu)

多線程代碼一個(gè)讓人比較頭疼的問(wèn)題，就是每個(gè)API我都懂，但就是不會(huì)用。很多對(duì)concurrent包倒背如流的同學(xué)，在面對(duì)現(xiàn)實(shí)的問(wèn)題時(shí)，到最后依然不得不被迫加上Lock或者synchronized。

ConcurrentBag是一個(gè)Lock free的數(shù)據(jù)結(jié)構(gòu)，主要用作數(shù)據(jù)庫(kù)連接的存儲(chǔ)，可以說(shuō)整個(gè)HikariCP的核心就是它。刪掉亂七八糟的注釋和異常處理，可以說(shuō)關(guān)鍵的代碼也就百十來(lái)行，但里面的道道卻非常的多。

ConcurrentBag速度很快，要達(dá)到這個(gè)目標(biāo)，就需要一定的核心數(shù)據(jù)結(jié)構(gòu)支持。

privatefinalCopyOnWriteArrayListsharedList;
privatefinalThreadLocal>threadList;
privatefinalAtomicIntegerwaiters;
privatefinalSynchronousQueuehandoffQueue;

• sharedList 用來(lái)緩存所有的連接，是一個(gè)CopyOnWriteArrayList結(jié)構(gòu)。
• threadList 用來(lái)緩存某個(gè)線程所使用的所有連接，相當(dāng)于快速引用，是一個(gè)ThreadLocal類型的ArrayList。
• waiters 當(dāng)前正在獲取連接的等待者數(shù)量。AtomicInteger，就是一個(gè)自增對(duì)象。當(dāng)waiters的數(shù)量大于0時(shí)候，意味著有線程正在獲取資源。
• handoffQueue 0容量的快速傳遞隊(duì)列，SynchronousQueue類型的隊(duì)列，非常有用。

ConcurrentBag里面的元素，為了能夠無(wú)鎖化操作，需要使用一些變量來(lái)標(biāo)識(shí)現(xiàn)在處于的狀態(tài)。抽象的接口如下：

publicinterfaceIConcurrentBagEntry{
intSTATE_NOT_IN_USE=0;
intSTATE_IN_USE=1;
intSTATE_REMOVED=-1;
intSTATE_RESERVED=-2;

booleancompareAndSet(intexpectState,intnewState);
voidsetState(intnewState);
intgetState();
}

有了這些數(shù)據(jù)結(jié)構(gòu)的支持，我們的ConcurrentBag就可以實(shí)現(xiàn)它光的宣稱了。

基于 Spring Boot + MyBatis Plus + Vue & Element 實(shí)現(xiàn)的后臺(tái)管理系統(tǒng) + 用戶小程序，支持 RBAC 動(dòng)態(tài)權(quán)限、多租戶、數(shù)據(jù)權(quán)限、工作流、三方登錄、支付、短信、商城等功能

• 項(xiàng)目地址：https://github.com/YunaiV/ruoyi-vue-pro
• 視頻教程：https://doc.iocoder.cn/video/

獲取連接

連接的獲取是borrow方法，還可以傳入一個(gè)timeout作為超時(shí)控制。

publicTborrow(longtimeout,finalTimeUnittimeUnit)throwsInterruptedException

首先，如果某個(gè)線程執(zhí)行非常快，使用了比較多的連接，就可以使用ThreadLocal的方式快速獲取連接對(duì)象，而不用跑到大池子里面去獲取。代碼如下。

//Trythethread-locallistfirst
finalvarlist=threadList.get();
for(inti=list.size()-1;i>=0;i--){
finalvarentry=list.remove(i);
finalTbagEntry=weakThreadLocals?((WeakReference)entry).get():(T)entry;
if(bagEntry!=null&&bagEntry.compareAndSet(STATE_NOT_IN_USE,STATE_IN_USE)){
returnbagEntry;
}
}

我們都知道，包括ArrayList和HashMap一些基礎(chǔ)的結(jié)構(gòu)，都是Fail Fast的，如果你在遍歷的時(shí)候，刪掉一些數(shù)據(jù)，有可能會(huì)引起問(wèn)題。幸運(yùn)的是，由于我們的List是從ThreadLocal獲取的，它首先就避免了線程安全的問(wèn)題。

接下來(lái)就是遍歷。這段代碼采用的是尾遍歷（頭遍歷會(huì)出現(xiàn)錯(cuò)誤），用于快速的從列表中找到一個(gè)可以復(fù)用的對(duì)象，然后使用CAS來(lái)把狀態(tài)置為使用中。但如果對(duì)象正在被使用，則直接刪除它。

在ConcurrentBag里，每個(gè)ThreadLocal最多緩存50個(gè)連接對(duì)象引用。

當(dāng)ThreadLocal里找不到可復(fù)用的對(duì)象，它就會(huì)到大池子里去拿。也就是下面這段代碼。

//Otherwise,scanthesharedlist...thenpollthehandoffqueue
finalintwaiting=waiters.incrementAndGet();
try{
for(TbagEntry:sharedList){
if(bagEntry.compareAndSet(STATE_NOT_IN_USE,STATE_IN_USE)){
//Ifwemayhavestolenanotherwaiter'sconnection,requestanotherbagadd.
if(waiting>1){
listener.addBagItem(waiting-1);
}
returnbagEntry;
}
}

listener.addBagItem(waiting);

//還拿不到，就需要等待別人釋放了
timeout=timeUnit.toNanos(timeout);
do{
finalvarstart=currentTime();
finalTbagEntry=handoffQueue.poll(timeout,NANOSECONDS);
if(bagEntry==null||bagEntry.compareAndSet(STATE_NOT_IN_USE,STATE_IN_USE)){
returnbagEntry;
}

timeout-=elapsedNanos(start);
}while(timeout>10_000);

returnnull;
}
finally{
waiters.decrementAndGet();
}

首先要注意，這段代碼可能是由不同的線程執(zhí)行的，所以必須要考慮線程安全問(wèn)題。由于shardList是線程安全的CopyOnWriteArrayList，適合讀多寫少的場(chǎng)景，我們可以直接進(jìn)行遍歷。

這段代碼的目的是一樣的，需要從sharedList找到一個(gè)空閑的連接對(duì)象。這里把自增的waiting變量傳遞到外面的代碼進(jìn)行處理，主要是由于想要根據(jù)waiting的大小來(lái)確定是否創(chuàng)建新的對(duì)象。

如果無(wú)法從池子里獲取連接，則需要等待別的線程釋放一些資源。

創(chuàng)建對(duì)象的過(guò)程是異步的，要想獲取它，還需要依賴一段循環(huán)代碼。while循環(huán)代碼是納秒精度，會(huì)嘗試從handoffQueue里獲取。最終會(huì)調(diào)用SynchronousQueue的transfer方法。

基于 Spring Cloud Alibaba + Gateway + Nacos + RocketMQ + Vue & Element 實(shí)現(xiàn)的后臺(tái)管理系統(tǒng) + 用戶小程序，支持 RBAC 動(dòng)態(tài)權(quán)限、多租戶、數(shù)據(jù)權(quán)限、工作流、三方登錄、支付、短信、商城等功能

• 項(xiàng)目地址：https://github.com/YunaiV/yudao-cloud
• 視頻教程：https://doc.iocoder.cn/video/

歸還連接

有借就有還，當(dāng)某個(gè)連接使用完畢，它將被歸還到池子中。

publicvoidrequite(finalTbagEntry)
{
bagEntry.setState(STATE_NOT_IN_USE);

for(vari=0;waiters.get()>0;i++){
if(bagEntry.getState()!=STATE_NOT_IN_USE||handoffQueue.offer(bagEntry)){
return;
}
elseif((i&0xff)==0xff){
parkNanos(MICROSECONDS.toNanos(10));
}
else{
Thread.yield();
}
}

finalvarthreadLocalList=threadList.get();
if(threadLocalList.size()50){
threadLocalList.add(weakThreadLocals?newWeakReference<>(bagEntry):bagEntry);
}
}

首先，把這個(gè)對(duì)象置為可用狀態(tài)。然后，代碼會(huì)進(jìn)入一個(gè)循環(huán)，等待使用方把這個(gè)連接接手過(guò)去。當(dāng)連接處于STATE_NOT_IN_USE狀態(tài)，或者隊(duì)列中的數(shù)據(jù)被取走了，那么就可以直接返回了。

由于waiters.get()是實(shí)時(shí)獲取的，有可能長(zhǎng)時(shí)間一直大于0，這樣代碼就會(huì)變成死循環(huán)，浪費(fèi)CPU。代碼會(huì)嘗試不同層次的睡眠，一個(gè)是每隔255個(gè)waiter睡10ns，一個(gè)是使用yield讓出cpu時(shí)間片。

如果歸還連接的時(shí)候并沒(méi)有被其他線程獲取到，那么最后我們會(huì)把歸還的連接放入到相對(duì)應(yīng)的ThreadLocal里，因?yàn)閷?duì)一個(gè)連接來(lái)說(shuō)，借和還，通常是一個(gè)線程。

知識(shí)點(diǎn)

看起來(lái)平平無(wú)奇的幾行代碼，為什么搞懂了就能Hold住大部分的并發(fā)編程場(chǎng)景呢？主要還是這里面的知識(shí)點(diǎn)太多。下面我簡(jiǎn)單羅列一下，你可以逐個(gè)攻破。

1. 使用ThreadLocal來(lái)緩存本地資源引用，使用線程封閉的資源來(lái)減少鎖的沖突
2. 采用讀多寫少的線程安全的CopyOnWriteArrayList來(lái)緩存所有對(duì)象，幾乎不影響讀取效率
3. 使用基于CAS的AtomicInteger來(lái)計(jì)算等待者的數(shù)量，無(wú)鎖操作使得計(jì)算更加快速
4. 0容量的交換隊(duì)列SynchronousQueue，使得對(duì)象傳遞更加迅速
5. 采用compareAndSet的CAS原語(yǔ)來(lái)控制狀態(tài)的變更，安全且效率高。很多核心代碼都是這么設(shè)計(jì)的
6. 在循環(huán)中使用park、yield等方法，避免死循環(huán)占用大量CPU
7. 需要了解并發(fā)數(shù)據(jù)結(jié)構(gòu)中的offer、poll、peek、put、take、add、remove方法的區(qū)別，并靈活應(yīng)用
8. CAS在設(shè)置狀態(tài)時(shí)，采用了volatile關(guān)鍵字修飾，對(duì)于volatile的使用也是一個(gè)常見(jiàn)的優(yōu)化點(diǎn)
9. 需要了解WeakReference弱引用在垃圾回收時(shí)候的表現(xiàn)

麻雀雖小，五臟俱全。如果你想要你的多線程編程能力更上一層樓，讀一讀這個(gè)短小精悍的ConcurrentBag吧。當(dāng)你掌握了它，多線程的那些東西，不過(guò)是小菜一碟。

審核編輯 ：李倩