Netty分布式高性能工具类recycler的使用及创建

2024-04-02 19:55

短信预约 -IT技能 免费直播动态提醒

recycler的使用

这一小节开始学习recycler相关的知识, recycler是netty实现的一个轻量级对象回收站, 在netty中, recycler的使用也是相当之频繁的

recycler作用是保证了对象的循环利用, 对象使用完可以通过recycler回收, 需要再次使用则从对象池中取出, 不用每次都创建新对象从而减少对系统资源的占用, 同时也减轻了gc的压力

这里看一个示例

public class RecyclerDemo {
    private static final Recycler<User> RECYCLER = new Recycler<User>() {
        @Override
        protected User newObject(Handle<User> handle) {
            return new User(handle);
        }
    };
    static class User{
        private final Recycler.Handle<User> handle;
        public User(Recycler.Handle<User> handle){
            this.handle=handle;
        }
        public void recycle(){
            handle.recycle(this);
        }
    }
    public static void main(String[] args){
        User user1 = RECYCLER.get();
        user1.recycle();
        User user2 = RECYCLER.get();
        user2.recycle();
        System.out.println(user1==user2);
    }
}

首先定义了一个Recycler的成员变量RECYCLER, 在匿名内部类中重写了newObject方法, 也就是创建对象的方法, 该方法就是用户自定义的

这里newObject返回的new User(handle), 代表当回收站没有此类对象的时候, 可以通过这种方式创建对象

成员变量RECYCLER, 可以用来对此类对象的回收和再利用

定一个了一个静态内部类User, User中有个成员变量handle, 在构造方法中为其赋值, handle的作用, 就是用于对象回收的

并且定义了一个方法recycle, 方法体中通过handle.recycle(this)这种方式将自身对象进行回收, 通过这步操作, 就可以将对象回收到Recycler中

以上逻辑先做了解, 之后会进行详细分析

在main方法中, 通过RECYCLER的get方法获取一个user, 然后进行回收

再通过get方法将回收站的对象取出, 再次进行回收, 最后判断两次取出的对象是否为一个对象, 最后结果输出为true

以上demo就可以说明Recycler的回收再利用的功能

简单介绍了demo, 我们就详细的分析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, 
                ratioMask, maxDelayedQueuesPerThread);
    }
};

这一段逻辑我们并不陌生, 在上一小节的学习中我们知道, 这里用于保存线程共享对象, 而这里的共享对象, 就是一个Stack类型的对象

每个stack中维护着一个DefaultHandle类型的数组, 用于盛放回收的对象, 有关stack和线程的关系如图所示:

8-3-1

也就是说在每个Recycler中, 都维护着一个线程共享的栈, 用于对一类对象的回收

跟到Stack的构造方法中

Stack(Recycler&lt;T&gt; parent, Thread thread, int maxCapacity, int maxSharedCapacityFactor, 
      int ratioMask, int maxDelayedQueues) {
    this.parent = parent;
    this.thread = thread;
    this.maxCapacity = maxCapacity;
    availableSharedCapacity = new AtomicInteger(max(maxCapacity / maxSharedCapacityFactor, LINK_CAPACITY));
    elements = new DefaultHandle[min(INITIAL_CAPACITY, maxCapacity)];
    this.ratioMask = ratioMask;
    this.maxDelayedQueues = maxDelayedQueues;
}

首先介绍几个构造方法中初始化的关键属性:

属性parent表示Reclycer对象自身

属性thread表示当前stack绑定的哪个线程

属性maxCapacity表示当前stack的最大容量, 表示stack最多能盛放多少个元素

属性elements, 就表示stack中存储的对象, 类型为DefaultHandle, 可以被外部对象引用, 从而实现回收

属性ratioMask是用来控制对象回收的频率的, 也就是说每次通过Reclycer回收对象的时候, 不是每次都会进行回收, 而是通过该参数控制回收频率

属性maxDelayedQueues, 这里稍微有些复杂, 在很多时候, 一个线程创建的对象, 有可能会被另一个线程所释放, 而另一个线程释放的对象是不会放在当前线程的stack中的, 而是会存放在一个叫做WeakOrderQueue的数据结构中, 里面也是存放着一个个DefaultHandle, WeakOrderQueue会存放线程1创建的并且在线程2进行释放的对象

这里只是稍作了解, 之后的会对此做详细剖析, 这里我们只需知道, maxDelayedQueues属性的意思就是我这个线程能回收几个其他创建的对象的线程, 假设当前线程是线程1, maxDelayedQueues为2, 那么我线程1回收了线程2创建的对象, 又回收了线程3创建的对象, 那么不可能回收线程4创建的对象了, 因为maxDelayedQueues2, 我只能回收两个线程创建的对象

属性availableSharedCapacity, 表示在线程1中创建的对象, 在其他线程中缓存的最大个数, 同样, 相关逻辑会在之后的内容进行剖析

另外介绍两个没有在构造方法初始化的属性:

private WeakOrderQueue cursor, prev;
private volatile WeakOrderQueue head;

这里相当于指针, 用于指向WeakOrderQueue的, 这里也是稍作了解, 之后会进行详细剖析

有关stack异线程之间对象的关系如图所示(简略):

8-3-2

我们再继续介绍Recycler的构造方法, 同时熟悉有关stack各个参数的默认值:

protected Recycler() {
    this(DEFAULT_MAX_CAPACITY_PER_THREAD);
}

这里调用了重载的构造方法, 并传入了参数DEFAULT_MAX_CAPACITY_PER_THREAD

DEFAULT_MAX_CAPACITY_PER_THREAD的默认值是32768, 在static块中被初始化的, 我们可以跟进去自行分析

这个值就代表的每个线程中, stack中最多回收的元素的个数

继续跟重载的构造方法

protected Recycler(int maxCapacityPerThread) {
    this(maxCapacityPerThread, MAX_SHARED_CAPACITY_FACTOR);
}

这里又调用了重载的构造方法, 并且传入刚才传入的32768和MAX_SHARED_CAPACITY_FACTOR

MAX_SHARED_CAPACITY_FACTOR默认值是2, 同样在static块中进行了初始化, 有关该属性的用处稍后讲解

继续跟构造方法:

protected Recycler(int maxCapacityPerThread, int maxSharedCapacityFactor) {
    this(maxCapacityPerThread, maxSharedCapacityFactor, RATIO, MAX_DELAYED_QUEUES_PER_THREAD);
}

这里同样调用了重载的构造方法, 传入了刚才32768和2, 还有两个属性RATIO和MAX_DELAYED_QUEUES_PER_THREAD

RATIO也在static中被初始化, 默认值是8

同上, MAX_DELAYED_QUEUES_PER_THREAD的默认值是2倍cpu核数

我们继续跟构造方法:

protected Recycler(int maxCapacityPerThread, int maxSharedCapacityFactor, 
                   int ratio, int maxDelayedQueuesPerThread) {
    ratioMask = safeFindNextPositivePowerOfTwo(ratio) - 1;
    if (maxCapacityPerThread &lt;= 0) {
        this.maxCapacityPerThread = 0;
        this.maxSharedCapacityFactor = 1;
        this.maxDelayedQueuesPerThread = 0;
    } else {
        this.maxCapacityPerThread = maxCapacityPerThread;
        this.maxSharedCapacityFactor = max(1, maxSharedCapacityFactor);
        this.maxDelayedQueuesPerThread = max(0, maxDelayedQueuesPerThread);
    }
}

这里将几个属性进行了初始化

首先看ratioMask, 这里的方法safeFindNextPositivePowerOfTwo的参数ratio为8, 该方法的意思就是大于等于8的2的幂次方-1, 这里就是ratioMask就是7

maxCapacityPerThread是刚才分析的32768, 是一个大于0的数, 所以进入else

maxCapacityPerThread为32768

maxSharedCapacityFactor的值为2

maxDelayedQueuesPerThread的值为2倍CPU核数

我们再回到Stack的构造方法中

Stack(Recycler&lt;T&gt; parent, Thread thread, int maxCapacity, int maxSharedCapacityFactor, 
      int ratioMask, int maxDelayedQueues) {
    this.parent = parent;
    this.thread = thread;
    this.maxCapacity = maxCapacity;
    availableSharedCapacity = new AtomicInteger(max(maxCapacity / maxSharedCapacityFactor, LINK_CAPACITY));
    elements = new DefaultHandle[min(INITIAL_CAPACITY, maxCapacity)];
    this.ratioMask = ratioMask;
    this.maxDelayedQueues = maxDelayedQueues;
}

根据Recycler初始化属性的逻辑, 我们可以知道Stack中几个属性的值: