当前位置 博文首页 > Netty源码解析 -- 对象池Recycler实现原理

    Netty源码解析 -- 对象池Recycler实现原理

    作者:binecy 时间:2021-01-09 12:02

    由于在Java中创建一个实例的消耗不小,很多框架为了提高性能都使用对象池,Netty也不例外。 本文主要分析Netty对象池Recycler的实现原理。

    由于在Java中创建一个实例的消耗不小,很多框架为了提高性能都使用对象池,Netty也不例外。
    本文主要分析Netty对象池Recycler的实现原理。

    源码分析基于Netty 4.1.52

    缓存对象管理

    Recycler的内部类Stack负责管理缓存对象。
    Stack关键字段

    // Stack所属主线程,注意这里使用了WeakReference
    WeakReference<Thread> threadRef;    
    // 主线程回收的对象
    DefaultHandle<?>[] elements;
    // elements最大长度
    int maxCapacity;
    // elements索引
    int size;
    // 非主线程回收的对象
    volatile WeakOrderQueue head;   
    

    Recycler将一个Stack划分给某个主线程,主线程直接从Stack#elements中存取对象,而非主线程回收对象则存入WeakOrderQueue中。
    threadRef字段使用了WeakReference,当主线程消亡后,该字段指向对象就可以被垃圾回收。

    DefaultHandle,对象的包装类,在Recycler中缓存的对象都会包装成DefaultHandle类。

    head指向的WeakOrderQueue,用于存放其他线程的对象

    WeakOrderQueue主要属性

    // Head#link指向Link链表首对象
    Head head;  
    // 指向Link链表尾对象
    Link tail;
    // 指向WeakOrderQueue链表下一对象
    WeakOrderQueue next;
    // 所属线程
    WeakReference<Thread> owner;
    

    Link中也有一个DefaultHandle<?>[] elements字段,负责存储数据。
    注意,Link继承了AtomicInteger,AtomicInteger的值存储elements的最新索引。

    WeakOrderQueue也是属于某个线程,并且WeakOrderQueue继承了WeakReference<Thread>,当所属线程消亡时,对应WeakOrderQueue也可以被垃圾回收。
    注意:每个WeakOrderQueue都只属于一个Stack,并且只属于一个非主线程。

    thread2要存放对象到Stack1中,只能存放在WeakOrderQueue1
    thread1要存放对象到Stack2中,只能存放在WeakOrderQueue3

    回收对象

    DefaultHandle#recycle -> Stack#push

    void push(DefaultHandle<?> item) {
        Thread currentThread = Thread.currentThread();
        if (threadRef.get() == currentThread) {
            // #1
            pushNow(item);
        } else {
            // #2
            pushLater(item, currentThread);
        }
    }
    

    #1 当前线程是主线程,直接将对象加入到Stack#elements中。
    #2 当前线程非主线程,需要将对象放到对应的WeakOrderQueue中

    private void pushLater(DefaultHandle<?> item, Thread thread) {
        ...
        // #1
        Map<Stack<?>, WeakOrderQueue> delayedRecycled = DELAYED_RECYCLED.get();
        WeakOrderQueue queue = delayedRecycled.get(this);
        if (queue == null) {
            // #2
            if (delayedRecycled.size() >= maxDelayedQueues) {
                delayedRecycled.put(this, WeakOrderQueue.DUMMY);
                return;
            }
            // #3
            if ((queue = newWeakOrderQueue(thread)) == null) {
                return;
            }
            delayedRecycled.put(this, queue);
        } else if (queue == WeakOrderQueue.DUMMY) {
            // #4
            return;
        }
        // #5
        queue.add(item);
    }
    

    #1 DELAYED_RECYCLED是一个FastThreadLocal,可以理解为Netty中的ThreadLocal优化类。它为每个线程维护了一个Map,存储每个Stack和对应WeakOrderQueue。
    所有这里获取的delayedRecycled变量是仅用于当前线程的。
    而delayedRecycled.get获取的WeakOrderQueue,是以Thread + Stack作为维度区分的,只能是一个线程操作。
    #2 当前WeakOrderQueue数量超出限制,添加WeakOrderQueue.DUMMY作为标记
    #3 构造一个WeakOrderQueue,加入到Stack#head指向的WeakOrderQueue链表中,并放入DELAYED_RECYCLED。这时是需要一下同步操作的。
    #4 遇到WeakOrderQueue.DUMMY标记对象,直接抛弃对象
    #5 将缓存对象添加到WeakOrderQueue中。

    WeakOrderQueue#add

    void add(DefaultHandle<?> handle) {
        handle.lastRecycledId = id;
    
        // #1
        if (handleRecycleCount < interval) {
            handleRecycleCount++;
            return;
        }
        handleRecycleCount = 0;
    
        
        Link tail = this.tail;
        int writeIndex;
        // #2
        if ((writeIndex = tail.get()) == LINK_CAPACITY) {
            Link link = head.newLink();
            if (link == null) {
                return;
            }
            this.tail = tail = tail.next = link;
            writeIndex = tail.get();
        }
        // #3
        tail.elements[writeIndex] = handle;
        handle.stack = null;
        // #4
        tail.lazySet(writeIndex + 1);
    }
    

    #1 控制回收频率,避免WeakOrderQueue增长过快。
    每8个对象都会抛弃7个,回收一个
    #2 当前Link#elements已全部使用,创建一个新的Link
    #3 存入缓存对象
    #4 延迟设置Link#elements的最新索引(Link继承了AtomicInteger),这样在该stack主线程通过该索引获取elements缓存对象时,保证elements中元素已经可见。

    获取对象

    Recycler#threadLocal中存放了每个线程对应的Stack。
    Recycler#get中首先获取属于当前线程的Stack,再从该Stack中获取对象,也就是,每个线程只能从自己的Stack中获取对象。
    Recycler#get -> Stack#pop

    DefaultHandle<T> pop() {
        int size = this.size;
        if (size == 0) {
            // #1
            if (!scavenge()) {
                return null;
            }
            size = this.size;
            if (size <= 0) {
                return null;
            }
        }
        // #2
        size --;
        DefaultHandle ret = elements[size];
        elements[size] = null;
        this.size = size;
    
        ...
        return ret;
    }
    

    #1 elements没有可用对象时,将WeakOrderQueue中的对象迁移到elements
    #2 从elements中取出一个缓存对象

    scavenge -> scavengeSome -> WeakOrderQueue#transfer

    boolean transfer(Stack<?> dst) {
        Link head = this.head.link;
        if (head == null) {
            return false;
        }
        // #1
        if (head.readIndex == LINK_CAPACITY) {
            if (head.next == null) {
                return false;
            }
            head = head.next;
            this.head.relink(head);
        }
        // #2
        final int srcStart = head.readIndex;
        int srcEnd = head.get();
        final int srcSize = srcEnd - srcStart;
        if (srcSize == 0) {
            return false;
        }
        // #3
        final int dstSize = dst.size;
        final int expectedCapacity = dstSize + srcSize;
    
        if (expectedCapacity > dst.elements.length) {
            final int actualCapacity = dst.increaseCapacity(expectedCapacity);
            srcEnd = min(srcStart + actualCapacity - dstSize, srcEnd);
        }
    
        if (srcStart != srcEnd) {
            final DefaultHandle[] srcElems = head.elements;
            final DefaultHandle[] dstElems = dst.elements;
            int newDstSize = dstSize;
            // #4
            for (int i = srcStart; i < srcEnd; i++) {
                DefaultHandle<?> element = srcElems[i];
                ...
                srcElems[i] = null;
                // #5
                if (dst.dropHandle(element)) {
                    continue;
                }
                element.stack = dst;
                dstElems[newDstSize ++] = element;
            }
            // #6
            if (srcEnd == LINK_CAPACITY && head.next != null) {
                this.head.relink(head.next);
            }
    
            head.readIndex = srcEnd;
            // #7
            if (dst.size == newDstSize) {
                return false;
            }
            dst.size = newDstSize;
            return true;
        } else {
            // The destination stack is full already.
            return false;
        }
    }
    

    就是把WeakOrderQueue中的对象迁移到Stack中。
    #1 head.readIndex 标志现在已迁移对象下标
    head.readIndex == LINK_CAPACITY,表示当前Link已全部移动,查找下一个Link
    #2 计算待迁移对象数量
    注意,Link继承了AtomicInteger
    #3 计算Stack#elements数组长度,不够则扩容
    #4 遍历待迁移的对象
    #5 控制回收频率
    #6 当前Link对象已全部移动,修改WeakOrderQueue#head的link属性,指向下一Link,这样前面的Link就可以被垃圾回收了。
    #7 dst.size == newDstSize 表示并没有对象移动,返回false
    否则更新dst.size

    其实对象池的实现难点在于线程安全。
    Recycler中将主线程和非主线程回收对象划分到不同的存储空间中(stack#elements和WeakOrderQueue.Link#elements),并且对于WeakOrderQueue.Link#elements,存取操作划分到两端进行(非主线程从尾端存入,主线程从首部开始读取),
    从而减少同步操作,并保证线程安全。

    另外,Netty还提供了更高级别的对象池类ObjectPool,使用方法可以参考PooledDirectByteBuf#RECYCLER,这里不再赘述。

    如果您觉得本文不错,欢迎关注我的微信公众号,系列文章持续更新中。您的关注是我坚持的动力!

    下一篇:没有了