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

时间 2020-11-23

标签数组缓存安全微信框架源码分析性能优化 this spa 栏目 Netty 繁體版

原文原文链接

因为在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中，只能存放在WeakOrderQueue3this

回收对象

DefaultHandle#recycle -> Stack#pushspa

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，这里再也不赘述。

若是您以为本文不错，欢迎关注个人微信公众号，系列文章持续更新中。您的关注是我坚持的动力！