窺探 Netty 源碼！Recycler 對(duì)象池實(shí)現(xiàn)原理剖析

本文選自 Doocs 開(kāi)源社區(qū)旗下“源碼獵人”項(xiàng)目，作者 tydhot。

項(xiàng)目將會(huì)持續(xù)更新，歡迎 Star 關(guān)注。

項(xiàng)目地址：https://github.com/doocs/source-code-hunter

該文所涉及的 netty 源碼版本為 4.1.6。

Netty 的對(duì)象池 Recycler 是什么

Recycler 是 Netty 中基于 ThreadLocal 的輕量化的對(duì)象池實(shí)現(xiàn)。既然是基于 ThreadLocal，那么就可以將其理解為當(dāng)前線程在通過(guò)對(duì)象池 Recycler 得到一個(gè)對(duì)象之后，在回收對(duì)象的時(shí)候，不需要將其銷(xiāo)毀，而是放回到該線程的對(duì)象池中即可，在該線程下一次用到該對(duì)象的時(shí)候，不需要重新申請(qǐng)空間創(chuàng)建，而是直接重新從對(duì)象池中獲取。

Recycler 在 netty 中被如何使用

Recycler 對(duì)象池在 netty 中最重要的使用，就在于 netty 的池化 ByteBuf 的場(chǎng)景下。首先，何為池化？以 PooledDirectByteBuf 舉例，每一個(gè) PooledDirectByteBuf 在應(yīng)用線程中使用完畢之后，并不會(huì)被釋放，而是等待被重新利用，類比線程池每個(gè)線程在執(zhí)行完畢之后不會(huì)被立即釋放，而是等待下一次執(zhí)行的時(shí)候被重新利用。所謂的對(duì)象池也是如此，池化減少了 ByteBuf 創(chuàng)建和銷(xiāo)毀的開(kāi)銷(xiāo)，也是 netty 高性能表現(xiàn)的基石之一。

private static final Recycler<PooledDirectByteBuf> RECYCLER = new Recycler<PooledDirectByteBuf>() {    @Override    protected PooledDirectByteBuf newObject(Handle<PooledDirectByteBuf> handle) {        return new PooledDirectByteBuf(handle, 0);    }};static PooledDirectByteBuf newInstance(int maxCapacity) {    PooledDirectByteBuf buf = RECYCLER.get();    buf.reuse(maxCapacity);    return buf;}

PooledDirectByteBuf 在其類加載的過(guò)程中，初始化了一個(gè)靜態(tài)的 RECYCLER 成員，通過(guò)重寫(xiě)其 newObject()方法達(dá)到使 Recycler 可以初始化一個(gè) PooledDirectByteBuf。而在接下來(lái)的使用中，只需要通過(guò)靜態(tài)方法 newInstance()就可以從 RECYCLER 對(duì)象池的 get()方法獲取一個(gè)新的 PooledDirectByteBuf 對(duì)象返回，而重寫(xiě)的方法 newObject()中的入?yún)?Handler 則提供了 recycle()方法給出了對(duì)象重新放入池中回收的能力，這里的具體實(shí)現(xiàn)在下文展開(kāi)。因此，newInstance()方法和 recycle()方法就提供了對(duì)象池出池和入池的能力，也通過(guò)此，PooledDirectByteBuf 達(dá)到了池化的目標(biāo)。

Recycler 的實(shí)現(xiàn)原理分析

Recycler 的實(shí)現(xiàn)原理很抽象，可以先直接閱讀文末的例子再閱讀這部分內(nèi)容。

Recycler 中，最核心的是兩個(gè)通過(guò) ThreadLocal 作為本地線程私有的兩個(gè)成員，而其實(shí)現(xiàn)原理只需要圍繞這兩個(gè)成員分析，就可以對(duì)對(duì)象池的設(shè)計(jì)有直接的理解和認(rèn)識(shí)。

?第一個(gè)成員是在 Recycler 被定義的 Stack 成員對(duì)象。

private final FastThreadLocal<Stack<T>> threadLocal = new FastThreadLocal<Stack<T>>() {    @Override    protected Stack<T> initialValue() {        return new Stack<T>(Recycler.this, Thread.currentThread(), maxCapacityPerThread, maxSharedCapacityFactor,                ratioMask, maxDelayedQueuesPerThread);    }};

顧名思義，這個(gè) Stack 主體是一個(gè)堆棧，但是其還維護(hù)著一個(gè)鏈表，而鏈表中的每一個(gè)節(jié)點(diǎn)都是一個(gè)隊(duì)列。

private DefaultHandle[] elements;private WeakOrderQueue cursor, prev;

上述的 elements 數(shù)組便是存放當(dāng)前線程被回收的對(duì)象，當(dāng)當(dāng)前線程從該線程的 Recycler 對(duì)象池嘗試獲取新的對(duì)象的時(shí)候，首先就會(huì)從當(dāng)前 Stack 的這個(gè)數(shù)組中嘗試獲取已經(jīng)在先前被創(chuàng)建并且在當(dāng)前線程被回收的對(duì)象，因?yàn)楫?dāng)對(duì)象池的對(duì)象在當(dāng)前線程被調(diào)用 recycle()的時(shí)候，是會(huì)直接放到 elements 數(shù)組中等待下一次的利用。那么問(wèn)題來(lái)了，如果從該線程中被申請(qǐng)的這個(gè)對(duì)象是在另外一個(gè)線程中被調(diào)用 recycle()方法回收呢？那么該對(duì)象就會(huì)處于鏈表中的隊(duì)列中，當(dāng)堆棧數(shù)組中的對(duì)象不存在的時(shí)候，將會(huì)嘗試把鏈表隊(duì)列中的對(duì)象轉(zhuǎn)移到數(shù)組中供當(dāng)前線程獲取。那么其他線程是如何把被回收的對(duì)象放到這些鏈表中的隊(duì)列的呢？接下來(lái)就是另一個(gè)成員的使命了。

?第二個(gè)成員是在 Recycler 中也是通過(guò) ThreadLocal 所實(shí)現(xiàn)的一個(gè)線程本地變量，DELAYED_RECYCLED ，是一個(gè) Stack 和隊(duì)列的映射 Map。

private static final FastThreadLocal<Map<Stack, WeakOrderQueue>> DELAYED_RECYCLED =        new FastThreadLocal<Map<Stack, WeakOrderQueue>>() {    @Override    protected Map<Stack, WeakOrderQueue> initialValue() {        return new WeakHashMap<Stack, WeakOrderQueue>();    }};

第二個(gè)成員 DELAYED_RECYCLED 可以通過(guò)上文的 Stack 獲取一個(gè)隊(duì)列。
在前一個(gè)成員的解釋中提到，當(dāng)別的線程調(diào)用另一個(gè)線程的對(duì)象池的 recycle()方法進(jìn)行回收的時(shí)候，并不會(huì)直接落到持有對(duì)象池的線程的 Stack 數(shù)組當(dāng)中，當(dāng)然原因也很簡(jiǎn)單，在并發(fā)情況下這樣的操作顯然是線程不安全的，而加鎖也會(huì)帶來(lái)性能的開(kāi)銷(xiāo)。因此，netty 在 Recycler 對(duì)象池中通過(guò)更巧妙的方式解決這一問(wèn)題。
在前面提到，除了數(shù)組，Stack 還持有了一系列隊(duì)列的組成的鏈表，這些鏈表中的每一個(gè)節(jié)點(diǎn)都是一個(gè)隊(duì)列，這些隊(duì)列又存放著別的線程所回收到當(dāng)前線程對(duì)象池的對(duì)象。那么，這些隊(duì)列就是各個(gè)線程針對(duì)持有對(duì)象池的專屬回收隊(duì)列，說(shuō)起來(lái)很拗口，看下面的代碼。

private void pushLater(DefaultHandle item, Thread thread) {    // we don't want to have a ref to the queue as the value in our weak map    // so we null it out; to ensure there are no races with restoring it later    // we impose a memory ordering here (no-op on x86)    Map<Stack, WeakOrderQueue> delayedRecycled = DELAYED_RECYCLED.get();    WeakOrderQueue queue = delayedRecycled.get(this);    if (queue == null) {        if (delayedRecycled.size() >= maxDelayedQueues) {            // Add a dummy queue so we know we should drop the object            delayedRecycled.put(this, WeakOrderQueue.DUMMY);            return;        }        // Check if we already reached the maximum number of delayed queues and if we can allocate at all.        if ((queue = WeakOrderQueue.allocate(this, thread)) == null) {            // drop object            return;        }        delayedRecycled.put(this, queue);    } else if (queue == WeakOrderQueue.DUMMY) {        // drop object        return;    }    queue.add(item);}private WeakOrderQueue(Stack stack, Thread thread) {    head = tail = new Link();    owner = new WeakReference<Thread>(thread);    synchronized (stack) {        next = stack.head;        stack.head = this;    }    // Its important that we not store the Stack itself in the WeakOrderQueue as the Stack also is used in    // the WeakHashMap as key. So just store the enclosed AtomicInteger which should allow to have the    // Stack itself GCed.    availableSharedCapacity = stack.availableSharedCapacity;}

pushLater()方法發(fā)生在當(dāng)一個(gè)對(duì)象被回收的時(shí)候，當(dāng)當(dāng)前線程不是這個(gè)對(duì)象所申請(qǐng)的時(shí)候的線程時(shí)，將會(huì)通過(guò)該對(duì)象的 Stack 直接去通過(guò) DELAYED_RECYCLED 映射到一條隊(duì)列上，如果沒(méi)有則創(chuàng)建并建立映射，再把該對(duì)象放入到該隊(duì)列中，以上操作結(jié)束后該次回收即宣告結(jié)束

private WeakOrderQueue(Stack stack, Thread thread) {    head = tail = new Link();    owner = new WeakReference<Thread>(thread);    synchronized (stack) {        next = stack.head;        stack.head = this;    }    // Its important that we not store the Stack itself in the WeakOrderQueue as the Stack also is used in    // the WeakHashMap as key. So just store the enclosed AtomicInteger which should allow to have the    // Stack itself GCed.    availableSharedCapacity = stack.availableSharedCapacity;}

如果在操作中，隊(duì)列是被創(chuàng)建的，會(huì)把該隊(duì)列放置在 Stack 中的鏈表里的頭結(jié)點(diǎn)，保證創(chuàng)建該對(duì)象的線程在數(shù)組空了之后能夠通過(guò)鏈表訪問(wèn)到該隊(duì)列并將該隊(duì)列中的回收對(duì)象重新放到數(shù)組中等待被下次重新利用，隊(duì)列交給 A 線程的鏈表是唯一的阻塞操作。在這里通過(guò)一次阻塞操作，避免后續(xù)都不存在資源的競(jìng)爭(zhēng)問(wèn)題。

舉一個(gè)例子來(lái)解釋對(duì)象池的原理

A 線程申請(qǐng)，A 線程回收的場(chǎng)景。

?顯然，當(dāng)對(duì)象的申請(qǐng)與回收是在一個(gè)線程中時(shí)，直接把對(duì)象放入到 A 線程的對(duì)象池中即可，不存在資源的競(jìng)爭(zhēng)，簡(jiǎn)單輕松。

A 線程申請(qǐng)，B 線程回收的場(chǎng)景。

?首先，當(dāng) A 線程通過(guò)其對(duì)象池申請(qǐng)了一個(gè)對(duì)象后，在 B 線程調(diào)用 recycle()方法回收該對(duì)象。顯然，該對(duì)象是應(yīng)該回收到 A 線程私有的對(duì)象池當(dāng)中的，不然，該對(duì)象池也失去了其意義。?那么 B 線程中，并不會(huì)直接將該對(duì)象放入到 A 線程的對(duì)象池中，如果這樣操作在多線程場(chǎng)景下存在資源的競(jìng)爭(zhēng)，只有增加性能的開(kāi)銷(xiāo)，才能保證并發(fā)情況下的線程安全，顯然不是 netty 想要看到的。?那么 B 線程會(huì)專門(mén)申請(qǐng)一個(gè)針對(duì) A 線程回收的專屬隊(duì)列，在首次創(chuàng)建的時(shí)候會(huì)將該隊(duì)列放入到 A 線程對(duì)象池的鏈表首節(jié)點(diǎn)（這里是唯一存在的資源競(jìng)爭(zhēng)場(chǎng)景，需要加鎖），并將被回收的對(duì)象放入到該專屬隊(duì)列中，宣告回收結(jié)束。?在 A 線程的對(duì)象池?cái)?shù)組耗盡之后，將會(huì)嘗試把各個(gè)別的線程針對(duì) A 線程的專屬隊(duì)列里的對(duì)象重新放入到對(duì)象池?cái)?shù)組中，以便下次繼續(xù)使用。

全文完！

希望本文對(duì)大家有所幫助。如果感覺(jué)本文有幫助，有勞轉(zhuǎn)發(fā)或點(diǎn)一下“在看”！讓更多人收獲知識(shí)！

推薦閱讀

?窺探 Netty 源碼！FastThreadLocal 究竟快在哪里？?牛逼！一文看懂 Java 并發(fā)編程在各主流框架中的應(yīng)用?把 AQS 源碼扒個(gè)「體無(wú)完膚」！

長(zhǎng)按識(shí)別下圖二維碼，關(guān)注公眾號(hào)「Doocs 開(kāi)源社區(qū)」，第一時(shí)間跟你們分享好玩、實(shí)用的技術(shù)文章與業(yè)內(nèi)最新資訊。