由于在 Java 中创建一个实例的消耗不小，很多框架为了提高性能都使用对象池，Netty 也不例外。
本文主要分析 Netty 对象池 Recycler 的实现原理。
源码分析基于 Netty 4.1.52
缓存对象管理
Recycler 的内部类 Stack 负责管理缓存对象。
Stack 关键字段
// Stack所属主线程，注意这里使用了WeakReferenceWeakReference<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，这里不再赘述。
如果您觉得本文不错，欢迎关注我的微信公众号，系列文章持续更新中。您的关注是我坚持的动力！
评论