1.對象池技術介紹

對象池其實就是緩存一些對象從而避免大量創(chuàng)建同一個類型的對象, 類似線程池。對象池緩存了一些已經(jīng)創(chuàng)建好的對象, 避免需要的時候創(chuàng)建。同時限制了實例的個數(shù)。

池化技術最終要的就是重復的使用池內(nèi)已經(jīng)創(chuàng)建的對象。

• 創(chuàng)建對象的開銷大
• 會創(chuàng)建大量的實例
• 限制一些資源的使用

如果創(chuàng)建一個對象的開銷特別大, 那么需要提前創(chuàng)建一些可以使用的并緩存起來, 池化技術就是重復使用對象, 提前創(chuàng)建并緩存起來重復使用就是池化, 可以降低創(chuàng)建對象的開銷。

會大量創(chuàng)建實例的場景, 重復的使用對象可以減少創(chuàng)建的對象數(shù)量, 降低GC的壓力。

對于限制資源的使用更多的是一種保護機制, 比如數(shù)據(jù)庫連接池, 除去創(chuàng)建對象本身的開銷, 們對外部系統(tǒng)也會造成壓力, 比如大量創(chuàng)建鏈接對DB也是有壓力的。那么池化除了可以優(yōu)化資源外, 本身限制了資源數(shù), 對外部系統(tǒng)也起到了一層保護作用。

2.如何實現(xiàn)對象池

Apache Commons Pool, Netty輕量級對象池的實現(xiàn)。

Apache Commons Pool開源軟件庫提供了一個對象池API和一系列對象池的實現(xiàn), 支持各種配置, 比如活躍對象數(shù)或者閑置對象個數(shù)等。DBCP數(shù)據(jù)庫連接池基于Apache Commons Pool實現(xiàn)。

Netty自己實現(xiàn)了一套輕量級的對象池, 在多個IO線程獨立工作時候, 基于NioEventLoop的實現(xiàn), 每個IO線程輪詢單獨的Selector實例來檢索IO事件, 在IO來臨時候開始處理。最常見的IO操作就是讀寫, 具體到NIO就是從內(nèi)核緩沖區(qū)拷貝數(shù)據(jù)到用戶緩沖區(qū)或者從用戶緩沖區(qū)拷貝數(shù)據(jù)到內(nèi)核緩沖區(qū)。

3.Netty對象池實現(xiàn)分析

NIO提供了兩種Buffer最為緩沖區(qū): DirectByteBuffer和HeapByteBuffer, Netty進行了池化提供性能。

Netty的池化分別為PooledDirectByteBuf 和PolledHeapByteBuf。

static PooledDirectByteBuf newInstance(int maxCapacity) {
        PooledDirectByteBuf buf = RECYCLER.get();
        buf.reuse(maxCapacity);
        return buf;
    }

可見RECYCLER是池化的核心, 通過RECYCLER.get獲取一個實例。

Recycler就是Netty實輕量級池化技術的核心。

public class RecycleTest {
    private static final Recycler<User> RECYCLER = new Recycler<User>(){
        @Override
        protected User newObject(Handle<User> handle) {
            return new User(handle);
        }
    };

    public static class User{
        private final Recycler.Handle<User> handle;

        public User(Recycler.Handle<User> handle){
            this.handle = handle;
        }

        public void recycle(){
            handle.recycle(this);
        }
    }

    public static void main(String[] args) {
        // 1.獲取當前線程的stack
        // 2.從stack中彈出對象
        // 3.創(chuàng)建對象并綁定到stack
        User user = RECYCLER.get();
        user.recycle();
        User user1 = RECYCLER.get();
        System.out.println(user == user1);
    }
}

true

3.1 Recycler

整個對象池的核心實現(xiàn)由ThreadLocal, Handler和Stack及WrakOrderQueue構成。

1. Stack相當于是一級緩存，同一個線程內(nèi)的使用和回收都將使用一個Stack
2. 每個線程都會有一個自己對應的Stack，如果回收的線程不是Stack的線程，將元素放入到Queue中
3. 所有的Queue組合成一個鏈表，Stack可以從這些鏈表中回收元素（實現(xiàn)了多線程之間共享回收的實例）

• get(): 獲取對象

    public final T get() {
        if (maxCapacityPerThread == 0) {
            return newObject((Handle<T>) NOOP_HANDLE);
        }
        Stack<T> stack = threadLocal.get();
        DefaultHandle<T> handle = stack.pop();
        if (handle == null) {
            handle = stack.newHandle();
            handle.value = newObject(handle);
        }
        return (T) handle.value;
    }

• 拿到當前線程對應的stack
• 從stack中pop出一個元素
• 如果不為空則返回，否則創(chuàng)建一個新的實例

    public final boolean recycle(T o, Handle<T> handle) {
        if (handle == NOOP_HANDLE) {
            return false;
        }

        DefaultHandle<T> h = (DefaultHandle<T>) handle;
        if (h.stack.parent != this) {
            return false;
        }

        h.recycle(o);
        return true;
    }

• Recycler的recycle方法主要做了一些參數(shù)驗證。

3.2 Handler

recycle(): 放回池子中的方法, Recycler的recycle方法和DefaultHandle的recycle方法。

        @Override
        public void recycle(Object object) {
            if (object != value) {
                throw new IllegalArgumentException("object does not belong to handle");
            }

            Stack<?> stack = this.stack;
            if (lastRecycledId != recycleId || stack == null) {
                throw new IllegalStateException("recycled already");
            }

            stack.push(this);
        }

• Recycler的recycle方法主要做了一些參數(shù)驗證。
• DefaultHandle的recycle方法:
  1. 如果當前線程是當前stack對象的線程， 放入stack。
  2. 獲取當前線程對應的Map<Stack, WeakOrderQueue>, 實例放入stack對應的queue中。

3.3 Stack

Stack是對象池化背后存儲實例的數(shù)據(jù)結構, 如果能從stack中拿到可用的實例就不再創(chuàng)建新的實例。

        final Recycler<T> parent;
        final WeakReference<Thread> threadRef;
        final AtomicInteger availableSharedCapacity;
        final int maxDelayedQueues;

        private final int maxCapacity;
        private final int ratioMask;
        private DefaultHandle<?>[] elements;
        private int size;
        private int handleRecycleCount = -1; // Start with -1 so the first one will be recycled.
        private WeakOrderQueue cursor, prev;
        private volatile WeakOrderQueue head;

pop()

        DefaultHandle<T> pop() {
            int size = this.size;
            // 如過size = 0, scavenge
            if (size == 0) {
                if (!scavenge()) {
                    return null;
                }
                size = this.size;
            }
            size --;
            DefaultHandle ret = elements[size];
            elements[size] = null;
            if (ret.lastRecycledId != ret.recycleId) {
                throw new IllegalStateException("recycled multiple times");
            }
            ret.recycleId = 0;
            ret.lastRecycledId = 0;
            this.size = size;
            // 返回elements最后一個元素
            return ret;
        }

push(): 同線程和異步線程回收對象

同步線程回收對象

        private void pushNow(DefaultHandle<?> item) {
            if ((item.recycleId | item.lastRecycledId) != 0) {
                throw new IllegalStateException("recycled already");
            }
            item.recycleId = item.lastRecycledId = OWN_THREAD_ID;

            int size = this.size;
            if (size >= maxCapacity || dropHandle(item)) {
                return;
            }
            if (size == elements.length) {
                elements = Arrays.copyOf(elements, min(size << 1, maxCapacity));
            }

            elements[size] = item;
            this.size = size + 1;
        }

異步線程回收對象

1. 獲取WeakOrderQueue
2. 創(chuàng)建WeakOrderQueue
3. 將對象追加到WeakOrderQueue

        private void pushLater(DefaultHandle<?> item, Thread thread) {
			// 獲取WeakOrderQueue
            Map<Stack<?>, WeakOrderQueue> delayedRecycled = DELAYED_RECYCLED.get();
            WeakOrderQueue queue = delayedRecycled.get(this);
            if (queue == null) {
                if (delayedRecycled.size() >= maxDelayedQueues) {
                    delayedRecycled.put(this, WeakOrderQueue.DUMMY);
                    return;
                }
                // 創(chuàng)建WeakOrderQueue
                if ((queue = WeakOrderQueue.allocate(this, thread)) == null) {
                    // drop object
                    return;
                }
                delayedRecycled.put(this, queue);
            } else if (queue == WeakOrderQueue.DUMMY) {
                // drop object
                return;
            }
			// 將對象追加到WeakOrderQueue
            queue.add(item);
        }

queue是一個隊列形式, 節(jié)點為Link, Link中是Handler (處理邏輯)

3.4 WeakOrderQueue

　　head,tail:Link                       // 內(nèi)部元素的指針（WeakOrderQueue內(nèi)部存儲的是一個Link的鏈表）
　　next:WeakOrderQueue     // 指向下一個WeakOrderQueue的指針
　　owner:Thread                      // 對應的線程

WeakOrderQueue的結構圖

WeakOrderQueue中是一個一個Link組成

add方法將元素添加到自身的“隊列”中, transfer方法將自己擁有的元素“傳輸”到Stack中。

        void add(DefaultHandle<?> handle) {
            handle.lastRecycledId = id;

            Link tail = this.tail;
            int writeIndex;
            if ((writeIndex = tail.get()) == LINK_CAPACITY) {
                if (!head.reserveSpace(LINK_CAPACITY)) {
                    // Drop it.
                    return;
                }
                // We allocate a Link so reserve the space
                this.tail = tail = tail.next = new Link();

                writeIndex = tail.get();
            }
            tail.elements[writeIndex] = handle;
            handle.stack = null;
            // we lazy set to ensure that setting stack to null appears before we unnull it in the owning thread;
            // this also means we guarantee visibility of an element in the queue if we see the index updated
            tail.lazySet(writeIndex + 1);
        }

add操作將元素添加到tail指向的Link對象中，如果Link已滿則創(chuàng)建一個新的Link實例。

3.5 Link

        static final class Link extends AtomicInteger {
        	// 處理邏輯Handler
            private final DefaultHandle<?>[] elements = new DefaultHandle[LINK_CAPACITY];
			// 讀的指針
            private int readIndex;
            Link next;
        }

Link內(nèi)部包含了一個數(shù)組用于存放實例, 同時標記了讀取位置的索引和下一個Link元素的指針。

4.總結

Recyler -> Stack-> WeakOrderQueue -> Link -> Handler

文章來源地址http://www.zghlxwxcb.cn/news/detail-417093.html

到了這里，關于[Netty源碼] Netty輕量級對象池實現(xiàn)分析 (十三)的文章就介紹完了。如果您還想了解更多內(nèi)容，請在右上角搜索TOY模板網(wǎng)以前的文章或繼續(xù)瀏覽下面的相關文章，希望大家以后多多支持TOY模板網(wǎng)！