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

如果您感觉本文不错，欢送关注我的微信公众号，系列文章继续更新中。您的关注是我保持的能源！