Netty源码分析第8章(高性能工具类FastThreadLocal和Recycler)—->第6节: 异线程回收对象
Netty源码分析第八章: 高性能工具类FastThreadLocal和Recycler
第六节: 异线程回收对象
异线程回收对象, 就是创建对象和回收对象不在同一条线程的情况下, 对象回收的逻辑
我们之前小节简单介绍过, 异线程回收对象, 是不会放在当前线程的stack中的, 而是放在一个WeakOrderQueue的数据结构中, 回顾我们之前的一个图:
8-6-1
相关的逻辑, 我们跟到源码中:
首先从回收对象的入口方法开始, DefualtHandle的recycle方法:
2
3
4
5
6
7
2 if (object != value) {
3 throw new IllegalArgumentException("object does not belong to handle");
4 }
5 stack.push(this);
6}
7
这部分我们并不陌生, 跟到push方法中:
2
3
4
5
6
7
8
9
2 Thread currentThread = Thread.currentThread();
3 if (thread == currentThread) {
4 pushNow(item);
5 } else {
6 pushLater(item, currentThread);
7 }
8}
9
上一小节分析过, 同线程会走到pushNow, 有关具体逻辑也进行了分析
如果不是同线程, 则会走到pushLater方法, 传入handle对象和当前线程对象
跟到pushLater方法中:
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
2 Map<Stack<?>, WeakOrderQueue> delayedRecycled = DELAYED_RECYCLED.get();
3 WeakOrderQueue queue = delayedRecycled.get(this);
4 if (queue == null) {
5 if (delayedRecycled.size() >= maxDelayedQueues) {
6 delayedRecycled.put(this, WeakOrderQueue.DUMMY);
7 return;
8 }
9 if ((queue = WeakOrderQueue.allocate(this, thread)) == null) {
10 return;
11 }
12 delayedRecycled.put(this, queue);
13 } else if (queue == WeakOrderQueue.DUMMY) {
14 return;
15 }
16 queue.add(item);
17}
18
首先通过DELAYED_RECYCLED.get()获取一个delayedRecycled对象
我们跟到DELAYED_RECYCLED中:
2
3
4
5
6
7
8
2 new FastThreadLocal<Map<Stack<?>, WeakOrderQueue>>() {
3 @Override
4 protected Map<Stack<?>, WeakOrderQueue> initialValue() {
5 return new WeakHashMap<Stack<?>, WeakOrderQueue>();
6 }
7};
8
这里我们看到DELAYED_RECYCLED是一个FastThreadLocal对象, initialValue方法创建一个WeakHashMap对象, WeakHashMap是一个map, key为stack, value为我们刚才提到过的WeakOrderQueue
从中我们可以分析到, 每个线程都维护了一个WeakHashMap对象
WeakHashMap中的元素, 是一个stack和WeakOrderQueue的映射, 说明了不同的stack, 对应不同的WeakOrderQueue
这里的映射关系可以举个例子说明:
比如线程1创建了一个对象, 在线程3进行了回收, 线程2创建了一个对象, 同样也在线程3进行了回收, 那么线程3对应的WeakHashMap中就会有两个元素:
线程1的stack和线程2的WeakOrderQueue, 线程2和stack和线程2的WeakOrderQueue
我们回到pushLater方法中:
继续往下看:
2
2
拿到了当前线程的WeakHashMap对象delayedRecycled之后, 然后通过delayedRecycled创建对象的线程的stack, 拿到WeakOrderQueue
这里的this, 就是创建对象的那个线程所属的stack, 这个stack是绑定在handle中的, 创建handle对象时候进行的绑定
假设当前线程是线程2, 创建handle的线程是线程1, 这里通过handle的stack拿到线程1的WeakOrderQueue
if (queue ==
null) 说明线程2没有回收过线程1的对象, 则进入if块的逻辑:
首先看判断
if (delayedRecycled.size() >= maxDelayedQueues)
delayedRecycled.size() 表示当前线程回收其他创建对象的线程的线程个数, 也就是有几个其他的线程在当前线程回收对象
maxDelayedQueues表示最多能回收的线程个数, 这里如果朝超过这个值, 就表示当前线程不能在回收其他线程的对象了
通过
delayedRecycled.put(
this, WeakOrderQueue.DUMMY) 标记, 创建对象的线程的stack, 所对应的WeakOrderQueue不可用, DUMMY我们可以理解为不可用
如果没有超过maxDelayedQueues, 则通过if判断中的
WeakOrderQueue.allocate(
this, thread) 这种方式创建一个WeakOrderQueue
allocate传入this, 也就是创建对象的线程对应的stack, 假设是线程1, thread就是当前线程, 假设是线程2
跟到allocate方法中:
2
3
4
5
2 return reserveSpace(stack.availableSharedCapacity, LINK_CAPACITY)
3 ? new WeakOrderQueue(stack, thread) : null;
4}
5
reserveSpace(stack.availableSharedCapacity, LINK_CAPACITY)表示线程1的stack还能不能分配LINK_CAPACITY个元素, 如果可以, 则直接通过new的方式创建一个WeakOrderQueue对象
再跟到reserveSpace方法中:
2
3
4
5
6
7
8
9
10
11
12
13
2 assert space >= 0;
3 for (;;) {
4 int available = availableSharedCapacity.get();
5 if (available < space) {
6 return false;
7 }
8 if (availableSharedCapacity.compareAndSet(available, available - space)) {
9 return true;
10 }
11 }
12}
13
参数availableSharedCapacity表示线程1的stack允许外部线程给其缓存多少个对象, 之前我们分析过是16384, space默认是16
方法中通过一个cas操作, 将16384减去16, 表示stack可以给其他线程缓存的对象数为16384-16
而这16个元素, 将由线程2缓存
回到pushLater方法中:
创建之后通过
delayedRecycled.put(
this, queue) 将stack和WeakOrderQueue进行关联
最后通过queue.add(item), 将创建的WeakOrderQueue添加一个handle
讲解WeakOrderQueue之前, 我们首先了解下WeakOrderQueue的数据结构
WeakOrderQueue维护了多个link, link之间是通过链表进行连接, 每个link可以盛放16个handle,
我们刚才分析过, 在reserveSpace方法中将
stack.availableSharedCapacity-16 , 其实就表示了先分配16个空间放在link里, 下次回收的时候, 如果这16空间没有填满, 则可以继续往里盛放
如果16个空间都已填满, 则通过继续添加link的方式继续分配16个空间用于盛放handle
WeakOrderQueue和WeakOrderQueue之间也是通过链表进行关联
可以根据下图理解上述逻辑:
8-6-2
根据以上思路, 我们跟到WeakOrderQueue的构造方法中:
2
3
4
5
6
7
8
9
10
2 head = tail = new Link();
3 owner = new WeakReference<Thread>(thread);
4 synchronized (stack) {
5 next = stack.head;
6 stack.head = this;
7 }
8 availableSharedCapacity = stack.availableSharedCapacity;
9}
10
这里有个head和tail, 都指向一个link对象, 这里我们可以分析到, 其实在WeakOrderQueue中维护了一个链表, head分别代表头结点和尾节点, 初始状态下, 头结点和尾节点都指向同一个节点
简单看下link的类的定义:
2
3
4
5
6
2 private final DefaultHandle<?>[] elements = new DefaultHandle[LINK_CAPACITY];
3 private int readIndex;
4 private Link next;
5}
6
每次创建一个Link, 都会创建一个DefaultHandle类型的数组用于盛放DefaultHandle对象, 默认大小是16个
readIndex是一个读指针, 我们之后小节会进行分析
next节点则指向下一个link
回到WeakOrderQueue的构造方法中:
owner是对向前线程进行一个包装, 代表了当前线程
接下来在一个同步块中, 将当前创建的WeakOrderQueue插入到stack指向的第一个WeakOrderQueue, 也就是stack的head属性, 指向我们创建的WeakOrderQueue, 如图所示
8-6-3
如果线程2创建一个和stack关联的WeakOrderQueue, stack的head节点就就会指向线程2创建WeakOrderQueue
如果之后线程3也创建了一个和stack关联的WeakOrderQueue, stack的head节点就会指向新创建的线程3的WeakOrderQueue
然后线程3的WeakOrderQueue再指向线程2的WeakOrderQueue
也就是无论哪个线程创建一个和同一个stack关联的WeakOrderQueue的时候, 都插入到stack指向的WeakOrderQueue列表的头部
这样就可以将stack和其他线程释放对象的容器WeakOrderQueue进行绑定
回到pushLater方法中:
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
2 Map<Stack<?>, WeakOrderQueue> delayedRecycled = DELAYED_RECYCLED.get();
3 WeakOrderQueue queue = delayedRecycled.get(this);
4 if (queue == null) {
5 if (delayedRecycled.size() >= maxDelayedQueues) {
6 delayedRecycled.put(this, WeakOrderQueue.DUMMY);
7 return;
8 }
9 if ((queue = WeakOrderQueue.allocate(this, thread)) == null) {
10 return;
11 }
12 delayedRecycled.put(this, queue);
13 } else if (queue == WeakOrderQueue.DUMMY) {
14 return;
15 }
16 queue.add(item);
17}
18
根据之前分析的WeakOrderQueue的数据结构, 我们分析最后一步, 也就是WeakOrderQueue的add方法
我们跟进WeakOrderQueue的add方法:
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
2 handle.lastRecycledId = id;
3 Link tail = this.tail;
4 int writeIndex;
5 if ((writeIndex = tail.get()) == LINK_CAPACITY) {
6 if (!reserveSpace(availableSharedCapacity, LINK_CAPACITY)) {
7 return;
8 }
9 this.tail = tail = tail.next = new Link();
10 writeIndex = tail.get();
11 }
12 tail.elements[writeIndex] = handle;
13 handle.stack = null;
14 tail.lazySet(writeIndex + 1);
15}
16
首先, 看
handle.lastRecycledId = id
lastRecycledId表示handle上次回收的id, 而id表示WeakOrderQueue的id, weakOrderQueue每次创建的时候, 会为自增一个唯一的id
Link tail =
this.tail 表示拿到当前WeakOrderQueue的中指向最后一个link的指针, 也就是尾指针
再看
if ((writeIndex = tail.get()) == LINK_CAPACITY)
tail.get()表示获取当前link中已经填充元素的个数, 如果等于16, 说明元素已经填充满
然后通过eserveSpace方法判断当前WeakOrderQueue是否还能缓存stack的对象, eserveSpace方法我们刚才已经分析过, 会根据stack的属性availableSharedCapacity-16的方式判断还能否缓存stack的对象, 如果不能再缓存stack的对象, 则返回
如果还能继续缓存, 则在创建一个link, 并将尾节点指向新创建的link, 并且原来尾节点的next的节点指向新创建的link
然后拿到当前link的writeIndex, 也就是写指针, 如果是新创建的link中没有元素, writeIndex为0
之后将尾部的link的elements属性, 也就是一个DefaultHandle类型的数组, 通过数组下标的方式将第writeIndex个节点赋值为要回收的handle
然后将handle的stack属性设置为null, 表示当前handle不是通过stack进行回收的
最后将tail节点的元素个数进行+1, 表示下一次将从writeIndex+1的位置往里写
以上就是异线程回收对象的逻辑
上一节: 同线程回收对象
下一节: 获取异线程释放的对象
posted on
2019-01-02 16:29 向南是个万人迷 阅读(
…) 评论(
…) 编辑 收藏
