Netty对象池-技术圈

在平时工作中,听说和使用过连接池,线程池等.还有一种就是对象池,可以实现对象复用的功能.

当然实现对象池的方式手段有多种,比如有一个公共的池子,所有需要对象的线程通过并发控制的方式从池子中获取对象,并发控制的同时伴随性能的损耗.那么Netty是如何实现对象池的呢? 先通过一段演示代码说起

import io.netty.util.Recycler;
public class Book {
    private String name;    private final Recycler.Handle<Book> recyclerHandle;
    Book(Recycler.Handle<Book> 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 Recycler<Book> BOOK = new Recycler<Book>() {        @Override        protected Book newObject(Handle<Book> 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(C),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<Stack<T>> threadLocal = new FastThreadLocal<Stack<T>>() {    @Override    protected Stack<T> initialValue() {        return new Stack<T>(Recycler.this, Thread.currentThread(), maxCapacityPerThread, maxSharedCapacityFactor,                            interval, maxDelayedQueuesPerThread, delayedQueueInterval);    }
    @Override    protected void onRemoval(Stack<T> 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 Recycler<Book> BOOK_CN = new Recycler<Book>() {        @Override        protected Book newObject(Handle<Book> handle) {            return new Book(handle);        }    };
    private static final Recycler<Book> BOOK_EN = new Recycler<Book>() {        @Override        protected Book newObject(Handle<Book> 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<T>) NOOP_HANDLE);    }    // 每个线程获取它自己对应的Stack对象.    Stack<T> stack = threadLocal.get();    // 从Stack中'弹出'一个对象.    DefaultHandle<T> 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中'弹出'一个对象的.

DefaultHandle<T> pop() {    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.Handle<Book> recyclerHandle;
    Book(Recycler.Handle<Book> 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<Stack<?>, 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获取元素的时候,代码如下

DefaultHandle<T> pop() {    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中.