netty源码阅读之性能优化工具类之Recycle异线程回收对象

水深无声 2022-05-08 01:34 119阅读 0赞

异线程回收对象分为以下几点分析：

1、获取WeakOrderQueue

2、如果WeakOrderQueue获取不到，说明是第一，那就创建WeakOrderQueue

3、获取到或者创建完WeakOrderQueue之后，将对象追加到WeakOrderQueue里面

我们从这个push开始看代码：

void push(DefaultHandle<?> item) {
                Thread currentThread = Thread.currentThread();
                if (thread == currentThread) {
                    // The current Thread is the thread that belongs to the Stack, we can try to push the object now.
                    pushNow(item);
                } else {
                    // The current Thread is not the one that belongs to the Stack, we need to signal that the push
                    // happens later.
                    pushLater(item, currentThread);
                }
            }

上一篇文章我们分析了pushNow(item)，现在我们是异线程，看pushLater方法：

private void pushLater(DefaultHandle<?> item, Thread thread) {
                // we don't want to have a ref to the queue as the value in our weak map
                // so we null it out; to ensure there are no races with restoring it later
                // we impose a memory ordering here (no-op on x86)
                Map<Stack<?>, WeakOrderQueue> delayedRecycled = DELAYED_RECYCLED.get();
                WeakOrderQueue queue = delayedRecycled.get(this);
                if (queue == null) {
                    if (delayedRecycled.size() >= maxDelayedQueues) {
                        // Add a dummy queue so we know we should drop the object
                        delayedRecycled.put(this, WeakOrderQueue.DUMMY);
                        return;
                    }
                    // Check if we already reached the maximum number of delayed queues and if we can allocate at all.
                    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);
            }

### 一、获取WeakOrderQueue  ###

我们先看DELAYED\_RECYCLED是什么：

private static final FastThreadLocal<Map<Stack<?>, WeakOrderQueue>> DELAYED_RECYCLED =
                new FastThreadLocal<Map<Stack<?>, WeakOrderQueue>>() {
            @Override
            protected Map<Stack<?>, WeakOrderQueue> initialValue() {
                return new WeakHashMap<Stack<?>, WeakOrderQueue>();
            }
        };

它是一个FastThreadLocal，值是Map，map的key是stack，value是WeakOrderQueue。这里表示每个线程都有一个map，这个map里面，不同的stack对应了不同的WeakOrderQueue。

然后继续看：

WeakOrderQueue queue = delayedRecycled.get(this);

假设我们有对象在A线程里面创建，然后在B线程里面回收，现在是B线程，获得到delayedRecycled之后，是一个map，这个map是线程B的，通过this（这个this是线程A的stack）就能获取到线程A的WeakOrderQueue。

### 二、那就创建WeakOrderQueue ###

看下面的这一段：

if (queue == null) {

也就是当获取到的WeakOrderQueue为空的时候，就要创建WeakOrderQueue了。

首先看第一个判断：

if (delayedRecycled.size() >= maxDelayedQueues) {
                        // Add a dummy queue so we know we should drop the object
                        delayedRecycled.put(this, WeakOrderQueue.DUMMY);
                        return;
                    }

这一段表示，delayedRecycle的长度已经大于maxDelayedQueues，也就是不能再回收别的线程的对象了，就传入WeakOrderQueue是DUMMY，DUMMY表示是假的意思，然后返回。这里和下面的这段代码对应：

else if (queue == WeakOrderQueue.DUMMY) {
                    // drop object
                    return;
                }

如果是DUMMY那就什么都不做。

接下去就是创建queue了：

// Check if we already reached the maximum number of delayed queues and if we can allocate at all.
                    if ((queue = WeakOrderQueue.allocate(this, thread)) == null) {
                        // drop object
                        return;
                    }

继续看allocate(this,thread)：

/**
             * Allocate a new {@link WeakOrderQueue} or return {@code null} if not possible.
             */
            static WeakOrderQueue allocate(Stack<?> stack, Thread thread) {
                // We allocated a Link so reserve the space
                return reserveSpace(stack.availableSharedCapacity, LINK_CAPACITY)
                        ? new WeakOrderQueue(stack, thread) : null;
            }

这里就是为thread（当然不是当前线程了，是stack的线程）创建stack的WeakOrderQueue。

这个三目运算符的意思就是，当前stack剩余的availableSharedCapacity，是否足以创建LINK\_CAPACITY的空间大小，如果足够就通过new WeakOrderQueue分配一个并返回，否则，返回null；

默认情况下availableSharedCapacity是16k，LINK\_CAPACITY是16，之前文章也是分析过。

那么我们再进入reserveSapce里面：

private static boolean reserveSpace(AtomicInteger availableSharedCapacity, int space) {
                assert space >= 0;
                for (;;) {
                    int available = availableSharedCapacity.get();
                    if (available < space) {
                        return false;
                    }
                    if (availableSharedCapacity.compareAndSet(available, available - space)) {
                        return true;
                    }
                }
            }

通过cas操作修改availableSharedCapacity，就是减去16。

通过这个操作之后，availableSharedCapacity=availableSharedCapacity-16。

返回，我们看看new WeakOrderQueue里面做了什么东西：

private WeakOrderQueue(Stack<?> stack, Thread thread) {
                head = tail = new Link();
                owner = new WeakReference<Thread>(thread);
                synchronized (stack) {
                    next = stack.head;
                    stack.head = this;
                }
    
                // Its important that we not store the Stack itself in the WeakOrderQueue as the Stack also is used in
                // the WeakHashMap as key. So just store the enclosed AtomicInteger which should allow to have the
                // Stack itself GCed.
                availableSharedCapacity = stack.availableSharedCapacity;
            }

我们首先简单地画个图表示一下WeakOrderQueue的数据结构：  
![70][]

WeakOrderQueue由一个个Link组成，有Head和Tail节点，最后还有一个指向别的WeakOrderQueue的next指针，也就是别的线程的WeakOrderQueue。

而Link又是由一个个handle组成的。LInk的默认大小是16，也就是下面有16个handle

![70 1][]

那为什么WeakOrderQueue数据结构里面，不是一个handle而是一个Link，Link再联系到handle呢？因为我们下次再回收对象的时候，就不需要判断是否还有 Link大小的位置了，如果Link里面还有空的handle就直接分配，提高了效率。也就是，我分配的时候是批量分配的，下次我就不用再判断是否有空间了，不用每次都去判断是否还有空间。

那么继续看源代码吧：

head = tail = new Link();

这段代码的意思就是，新建一个Link，并且把head和tail都指向这个Link。

然后把当前线程thread执行一个弱引用owner：

owner = new WeakReference<Thread>(thread);

然后这里有一个同步块：

synchronized (stack) {
                    next = stack.head;
                    stack.head = this;
                }

首先我们要理解，stack里面有多个WeakOrderQueue（对应多个线程的WeakOrderQueue），WeakOrderQueue里面又有多个Link，Link里面又有多个handle。

假设我们新建的WeakOrderQueue为w1，我们就把w1插入到stack1多个WeakOrderQueue 队列的头部，也就是：

w1->w2->w3。

### 三、将对象追加到WeakOrderQueue里面 ###

最后我们回来看这个方法：

queue.add(item);

那么继续看源码：

void add(DefaultHandle<?> handle) {
                handle.lastRecycledId = id;
    
                Link tail = this.tail;
                int writeIndex;
                if ((writeIndex = tail.get()) == LINK_CAPACITY) {
                    if (!reserveSpace(availableSharedCapacity, LINK_CAPACITY)) {
                        // Drop it.
                        return;
                    }
                    // We allocate a Link so reserve the space
                    this.tail = tail = tail.next = new Link();
    
                    writeIndex = tail.get();
                }
                tail.elements[writeIndex] = handle;
                handle.stack = null;
                // we lazy set to ensure that setting stack to null appears before we unnull it in the owning thread;
                // this also means we guarantee visibility of an element in the queue if we see the index updated
                tail.lazySet(writeIndex + 1);
            }

我们看handle.lastRecycledId=id，把id赋值给lastRecycledId，看id是什么：

private static final class WeakOrderQueue {
    
            ...
            private final int id = ID_GENERATOR.getAndIncrement();
            ...
    }

这个id就是WeakOrderQueue的id，也就表示，当前这个元素是这个WeakOrderQueue进行回收的。

然后通过(writeIndex = tail.get()) == LINK\_CAPACITY取出尾指针，看是否达到LINK\_CAPACITY也就是看是否里面的handle都写满了，如果没有，就又尝试分配一个Link追加到后面。我们看Link是什么东西：

// Let Link extend AtomicInteger for intrinsics. The Link itself will be used as writerIndex.
            @SuppressWarnings("serial")
            private static final class Link extends AtomicInteger {
                private final DefaultHandle<?>[] elements = new DefaultHandle[LINK_CAPACITY];
    
                private int readIndex;
                private Link next;
            }

继承自AtomicInteger，所以可以使用get，然后它里面的elements默认是LINK\_CAPACITY个，也就是16个。

所以下面writeIndex = tail.get();找出他的writeIndex，如果是新建的话就是首个位置0。

然后tail.elements\[writeIndex\] = handle;把handle传入。

由于这个handle已经没有stack了，它是WeakOrderQueue里面的，所以调用handle.stack = null;

最后把writeIndex+1，下次就从另外一个位置开始：

// we lazy set to ensure that setting stack to null appears before we unnull it in the owning thread;
                // this also means we guarantee visibility of an element in the queue if we see the index updated
                tail.lazySet(writeIndex + 1);

[70]: /images/20220508/134ddb168ebe44eb8f0b12e56d864f10.png
[70 1]: /images/20220508/dbc82c06c0954e598e0ba5ced32e0255.png