本文共 10605 字,大约阅读时间需要 35 分钟。
在平时工作中,听说和使用过连接池,线程池等.还有一种就是对象池,可以实现对象复用的功能.
当然实现对象池的方式手段有多种,比如有一个公共的池子,所有需要对象的线程通过并发控制的方式从池子中获取对象,并发控制的同时伴随性能的损耗.那么Netty是如何实现对象池的呢? 先通过一段演示代码说起
import io.netty.util.Recycler;public class Book { private String name; private final Recycler.HandlerecyclerHandle; Book(Recycler.Handle recyclerHandle) { this.recyclerHandle = recyclerHandle; } public void setName(String name) { this.name = name; } void recycle() { recyclerHandle.recycle(this); }}
import io.netty.util.Recycler;import io.netty.util.concurrent.FastThreadLocalThread;import java.util.concurrent.locks.LockSupport;public class Main { private static final RecyclerBOOK = new Recycler () { @Override protected Book newObject(Handle handle) { return new Book(handle); } }; private static Book book1; private static Book book2; private static Book book3; public static void main(String[] args) throws Exception { new FastThreadLocalThread(() -> { // book1,book2,book3都是从线程Thread-1中创建产生 book1 = BOOK.get(); book1.setName("Java"); book2 = BOOK.get(); book2.setName("C"); book3 = BOOK.get(); book3.setName("C++"); // book1在线程Thread-1中回收 book1.recycle(); new FastThreadLocalThread(() -> { // book2在线程Thread-2中回收 book2.recycle(); LockSupport.park();// 让线程不退出,便于观察 }, "Thread-2").start(); new FastThreadLocalThread(() -> { // book3在线程Thread-3中回收 book3.recycle(); LockSupport.park();// 让线程不退出,便于观察 }, "Thread-3").start(); LockSupport.park();// 让线程不退出,便于观察 }, "Thread-1").start(); // 让线程不退出,便于观察 LockSupport.park(); }}
以上代码,在线程Thread-1中创建了book1(Java),book2©,book3(C++)这三个对象.然后在线程Thread-1中回收book1对象,在线程Thread-2中回收book2对象,在线程Thread-3中回收book3对象.
通过jps命令查看进程ID
通过jmap命令dump堆内存
jmap -dump:format=b,file=2021-heap.hprof 10624
接下来使用Eclipse的MAT工具打开上面创建生成的2021-heap.hprof文件.
点击上图中的按钮,查看线程.
找到线程Thread-1的内存地址=0xd786b000
然后根据这个地址,查看线程内部属性信息.
线程Thread-1内部属性信息如下图所示
其中,需要关注threadLocalMap属性.
关于FastThreadLocal的内容,之前的文章有介绍.
这里也附着一张FastThreadLocal的图
代码中创建的book1,book2,book3这三个对象,就存放在threadLocalMap属性里面.依次展开它.
首先找到了book1对象
继续找到了book2对象,而且这个book2对象和线程Thread-2有关系.
最后找到了book3对象,而且这个book3对象和线程Thread-3有关系.
根据以上堆数据的分析,可以得出如下简单关系.
如上图,线程Thread-1创建了book1,book2,book3这三个对象. 回收之后,book1对象在线程Thread-1里,book2对象在线程Thread-2里,book3对象在线程Thread-3里.
接下来,通过分析源码的方式,完善这张图.
我们先将上面一开始的代码,简化下,如下图所示
根据上图,得到如下一个引用关系
Thread -> Recycler -> 目标对象
线程如果需要对象,不能在像以前那样手动创建对象(比如new Book()),而是需要借助Recycler对象,通过它得到所需要的Book对象.
因此,Recycler是一个很重要的类
io.netty.util.Recycler
在Recycler内部有一个很重要的属性,如下
private final FastThreadLocal> threadLocal = new FastThreadLocal >() { @Override protected Stack initialValue() { return new Stack (Recycler.this, Thread.currentThread(), maxCapacityPerThread, maxSharedCapacityFactor, interval, maxDelayedQueuesPerThread, delayedQueueInterval); } @Override protected void onRemoval(Stack value) { if (value.threadRef.get() == Thread.currentThread()) { if (DELAYED_RECYCLED.isSet()) { DELAYED_RECYCLED.get().remove(value); } } }};
通过线程局部对象FastThreadLocal, 也就是说, 当每个线程在使用Recycler获取所需要的对象的时候,它的内部是实际上是从每个线程的Stack中获取对象.
每个线程都会有一个Stack. 当然这句话也不是这么绝对, 如果一个线程它使用了n个Recycler,那么这个线程就拥有n个Stack. 为了说明这一点,验证代码如下
如果在一个线程中,使用多个Recycler对象.
import io.netty.util.Recycler;import io.netty.util.concurrent.FastThreadLocalThread;public class Main { private static final RecyclerBOOK_CN = new Recycler () { @Override protected Book newObject(Handle handle) { return new Book(handle); } }; private static final Recycler BOOK_EN = new Recycler () { @Override protected Book newObject(Handle handle) { return new Book(handle); } }; public static void main(String[] args) throws Exception { new FastThreadLocalThread(() -> { // 在同一个线程中使用2个Recycler实例,则会创建2个Stack Book book1 = BOOK_CN.get(); Book book2 = BOOK_EN.get(); }, "Thread-1").start(); }}
如上图,线程Thread-1它拥有2个Stack对象.
接下来继续分析Netty的对象池, 代码中是通过io.netty.util.Recycler#get方法获取对象的,追踪此方法.
public final T get() { // 如果没有启用线程池,则每次获取对象都要新创建对象. if (maxCapacityPerThread == 0) { return newObject((Handle) NOOP_HANDLE); } // 每个线程获取它自己对应的Stack对象. Stack stack = threadLocal.get(); // 从Stack中'弹出'一个对象. DefaultHandle handle = stack.pop(); if (handle == null) { // 如果Stack中没有DefaultHandle对象,则新创建一个DefaultHandle handle = stack.newHandle(); // 创建线程需要的那个对象 handle.value = newObject(handle); } // 返回线程需要的那个对象 return (T) handle.value;}
根据以上代码,得出如下关系.
Thread -> Recycler -> Stack -> DefaultHandle -> 目标对象
一个线程,要想得到所需要的目标对象,需要经过Recycler->Stack->DefaultHandle之后,才能拿到目标对象.
之前dump出来的堆内存,也能看到目标对象是’包裹’在DefaultHandle对象中的.
接下来看下它是怎么从Stack中’弹出’一个对象的.
DefaultHandlepop() { int size = this.size; if (size == 0) { // 这个地方后面会说 if (!scavenge()) { return null; } size = this.size; if (size <= 0) { return null; } } size --; // 从elements数组中拿出最后一个元素 DefaultHandle ret = elements[size]; elements[size] = null; this.size = size; if (ret.lastRecycledId != ret.recycleId) { throw new IllegalStateException("recycled multiple times"); } ret.recycleId = 0; ret.lastRecycledId = 0; // 将元素直接返回 return ret;}
在Stack内部有个数组,用来’装’DefaultHandle,当需要的时候,直接从这个数组中拿出最后一个元素DefaultHandle返回.
到目前为止,看一下此时的结构
正所谓’有借有还’,既然它是一个对象池,当使用完之后,需要调用回收方法. 在文章一开始我们自己设计的Book类中也实现了回收方法.
public class Book { private String name; private final Recycler.HandlerecyclerHandle; Book(Recycler.Handle recyclerHandle) { this.recyclerHandle = recyclerHandle; } public void setName(String name) { this.name = name; } // 回收方法 void recycle() { // 实际调用io.netty.util.Recycler.DefaultHandle#recycle方法 recyclerHandle.recycle(this); }}
回收操作就是从这个io.netty.util.Recycler.DefaultHandle#recycle方法开始的.
@Overridepublic void recycle(Object object) { if (object != value) { throw new IllegalArgumentException("object does not belong to handle"); } // 得到相应的Stack // 每个DefaultHandle有且仅属于一个Stack Stack stack = this.stack; if (lastRecycledId != recycleId || stack == null) { throw new IllegalStateException("recycled already"); } // 将DefaultHandle对象放入Stack中 stack.push(this);}
还要说一点是,如下图,比如线程Thread-1创建一个book对象,第一种情况,book对象使用完之后,最后是由线程Thread-1回收它的(谁创建谁回收). 第二种情况,book对象交给了线程Thread-3使用,最后由线程Thread-3回收它(他人创建我来回收).
根据上图,再理解下push源码
void push(DefaultHandle item) { // 得到当前线程 Thread currentThread = Thread.currentThread(); // 如果当前线程和Stack对应的线程是同一个线程 // 每个Stack有且只属于一个线程. if (threadRef.get() == currentThread) { pushNow(item); } else { // 当前线程和Stack对应的线程不是同一个线程 pushLater(item, currentThread); }}
将对象放入Stack分2种情况,第一种是pushNow,第二种是pushLater.
private void pushNow(DefaultHandle item) { if (item.recycleId != 0 || !item.compareAndSetLastRecycledId(0, OWN_THREAD_ID)) { throw new IllegalStateException("recycled already"); } item.recycleId = OWN_THREAD_ID; int size = this.size; // dropHandle方法用来控制放入Stack的速率,有点类似流控,稍后再说 if (size >= maxCapacity || dropHandle(item)) { return; } if (size == elements.length) { elements = Arrays.copyOf(elements, min(size << 1, maxCapacity)); } // 直接将元素放入elements数组的最后 elements[size] = item; this.size = size + 1;}
pushNow方法很简单,直接将DefaultHandle元素放入到数组即可.
pushLater方法有点麻烦.首先你要记得,执行pushLater的线程一定不是归属Stack的线程.
说白了,就是由其他线程’协助’归属Stack的线程来做push操作.
// 1.先将Queue链接到链表上// 2.将item添加到Queue#Linkprivate void pushLater(DefaultHandle item, Thread thread) { // DELAYED_RECYCLED是FastThreadLocal类型, 则每个线程都有一个DELAYED_RECYCLED. // 每个线程会维护 Stack -> WeakOrderQueue 映射关系, 即每个线程会'协助'其他线程存储它的对象. Map, WeakOrderQueue> delayedRecycled = DELAYED_RECYCLED.get(); WeakOrderQueue queue = delayedRecycled.get(this); if (queue == null) { if (delayedRecycled.size() >= maxDelayedQueues) { delayedRecycled.put(this, WeakOrderQueue.DUMMY); return; } // 创建queue,同时链接到链表上 if ((queue = WeakOrderQueue.allocate(this, thread)) == null) { // drop object return; } delayedRecycled.put(this, queue); } else if (queue == WeakOrderQueue.DUMMY) { // drop object return; } queue.add(item);}
结合上图,再梳理下pushLater逻辑
1.首先线程Thread-3先从Map中查找对应Stack的Queue. 第一次肯定找不到,于是新建一个Queue,然后再把这个新建的Queue链接到Stack上.
2.将元素添加到Queue.
到了现在,我们可以放一张全局图了
如上图,Stack对象是隶属于线程thread-1的. 如果从Stack中获取的对象,最后也是由线程thread-1回收的,那么对象就会存到Stack中的elements数组中.
如果从Stack中获取的对象,最后是由线程thread-2回收,那么thread-2就会创建一个针对此Stack的Queue,链接到Queue链上.然后再把对象放到相应的Link中的elements数组中.
还有一点,在回收对象的时候,并不是’全部都回收’. 而是默认每隔8个回收一个对象. 在pushNow和pushLater方法内部都会调用如下方法
boolean dropHandle(DefaultHandle handle) { // 1.如果当前handle之前被回收过,那么此次也会被回收 // 2.如果当前handle之前没有回收过,那么默认每隔8个回收一个,防止Stack的DefaultHandle[]数组发生爆炸性的增长. if (!handle.hasBeenRecycled) { if ((++handleRecycleCount & ratioMask) != 0) { // Drop the object. return true; } handle.hasBeenRecycled = true; // 标记元素被回收 } return false;}
在从Stack获取元素的时候,代码如下
DefaultHandlepop() { int size = this.size; if (size == 0) { // 如果elements中没有可用元素 if (!scavenge()) { // 从其他WeakOrderQueue中转移数据到当前stack#elements中 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; return ret;}
默认先从Stack自己的elements中获取元素,如果elements中没有元素的时候,则从Queue链上转移数据到当前Stack的elements数组中.
分析到这里,我们可以总结下Netty对象池的实现了.
每个线程都有一个Stack用于’装载’需要复用的对象. 同时其他线程也会’协助’它回收对象. 如果Stack中没有对象了,那么会从其他线程的Queue中转移对象到Stack中. 如果是Stack隶属的线程回收对象,那么对象会被放到Stack的elements数组中,如果是其他线程回收对象,那么会把这个对象放到其他线程维护的Queue中.公众号
转载地址:http://ypcki.baihongyu.com/