AbstractQueuedSynchronizer---condition队列的await方法中为什么要释放锁（JDK1.8）

红太狼 2023-09-27 09:33 5阅读 0赞

#### 目录 ####

*  **前言**
 *  **ArrayBlockingQueue的一些基础概念**
 *   *  **put**
     *  **take**
 *  **ConditionObject的wait方法**
 *  **ConditionObject的signal方法**
 *  end

## **前言** ##

AbstractQueuedSynchronizer的ConditionObject会维护一个单链表构建的条件队列。直接在AbstractQueuedSynchronizer中看ConditionObject的await源码，就会让人很迷惑。在node加入condition队列之后会释放锁。就这个释放锁的操作，只看await方法是无论如何也想不到为什么要释放锁，这个操作看起来莫名其妙的。

为了看懂这个部分，还需要从AQS条件队列的应用来理解。AQS的条件队列可以应用于 ArrayBlockingQueue，从ArrayBlockingQueue来看AQS的await方法就很好理解了。下面会先简单的介绍ArrayBlockingQueue。

ArrayBlockingQueue是一个线程安全的队列，添加元素的方法有三个：add，offer，put；

*  add：调用父类的add方法，最终还是会调用offer方法；添加元素成功返回true，失败抛异常；不会阻塞线程
 *  offer：添加元素成功返回true，失败返回false，不会阻塞线程；
 *  put：添加元素失败：阻塞线程；成功：不阻塞线程，没有返回值；

获取元素的方法：

*  take：获取不到元素，阻塞线程；
 *  poll：获取不到，返回null；不阻塞线程
 *  peek：返回一个元素对象，不会将元素从数组中清除，不阻塞线程；

本文主要通过ArrayBlockingQueue的put，take方法来分析条件队列的用法，通过对条件队列的用法的分析来理解AbstractQueuedSynchronizer中await，signal方法；

## **ArrayBlockingQueue的一些基础概念** ##

ArrayBlockingQueue是一个定长数组结构，所有的元素存放在数组中，在初始化ArrayBlockingQueue对象时，必须要指定数组的长度。指定数组长度之后，这个值不会更改。因为这个性质，才会有notFull队列。

ArrayBlockingQueue是线程安全的，适用于多线程环境。ArrayList与ArrayBlockingQueue的数据结构类似，都是用数组来存储元素。ArrayList不是线程安全的，那如果用ArrayList在多线程环境下进行元素的存入，取出为什么不安全呢？而ArrayBlockingQueue又是如何保证元素的存入，取出都是线程安全的呢？

![在这里插入图片描述][bc7219cda3e342009ddf3366d3fd1ae9.png_pic_center]  
  在多线程环境下，必须要考虑到多个线程同时操作一个位置上的数据，如果使用ArrayList添加元素，那么可能多个线程同时将数据添加到同一个位置上，这样会造成多个线程的数据被覆盖，只有一个线程添加成功；同样的取出数据的时候，也可能是多个线程获取到同一个位置的同一个对象；以数据库连接来看，我们希望的是每个线程可以获取不同的连接对象，而不是多个线程共用一个对象；

**为了在多线程环境下的数据安全ArrayBlockingQueue在获取，添加元素时都会先获取锁。这样可以保证多线程环境下同一个位置的元素添加，获取都只有一个线程操作，这种操作是线程安全的。**

在ArrayBlockingQueue中有2个条件队列：notFull，notEmpty；什么条件使用这2个队列呢？

*  notFull ：当数组被元素填满之后，还有线程要往ArrayBlockingQueue中添加元素时，由于数组长度不变，因此这些线程要等数组中有空出来的位置时，才能往数组中添加元素。在等待期间，线程就被放在notFull的条件队列中。  
    ![在这里插入图片描述][3faf33008b4c4663bd23fdc005395b52.png_pic_center]
 *  notEmpty：在获取元素时如果数组中没有元素，就只能等待数组中有元素才能从数组中获取；这些等待获取元素的线程就被放在notEmpty队列中；![在这里插入图片描述][891fa33acb174651a179c8cdf5143742.png_pic_center]

下面分别分析ArrayBlockingQueue的put，take方法

### **put** ###

源码：

public void put(E e) throws InterruptedException {
        
            checkNotNull(e);
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly();
            try {
        
                while (count == items.length)
                    notFull.await();
                enqueue(e);
            } finally {
        
                lock.unlock();
            }
        }
    
        private void enqueue(E x) {
        
            // assert lock.getHoldCount() == 1;
            // assert items[putIndex] == null;
            final Object[] items = this.items;
            items[putIndex] = x;
            if (++putIndex == items.length)
                putIndex = 0;
            count++;
            notEmpty.signal();
        }

put方法的源码很简单，首先为了保证线程安全，要获取锁；再检查数组是否被元素填满。如果被填满，就需要将线程添加到notFull队列中等待了；如果数组有空位置，就直接往数组添加元素；往数组添加元素成功之后，最后需要从notEmpty中唤醒线程。

那为什么要添加元素之后，最后要从notEmpty中唤醒线程呢？

试想一下这样的场景：在数组中没有元素时，已经有线程调用take方法从数组获取元素。由于这个时候数组中没有元素，因此只能将获取元素的线程添加到notEmpty队列中等待获取元素。put添加元素之后就唤醒notEmpty队列中的线程，保证了在数组有元素时，notEmpty中等待获取元素的线程能够获取到数组的元素。而不会发生数组有元素但notEmpty队列中的线程却不能获取到元素的情况。

### **take** ###

源码：

public E take() throws InterruptedException {
        
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly();
            try {
        
                while (count == 0)
                    notEmpty.await();
                return dequeue();
            } finally {
        
                lock.unlock();
            }
        }
    
        private E dequeue() {
        
            // assert lock.getHoldCount() == 1;
            // assert items[takeIndex] != null;
            final Object[] items = this.items;
            @SuppressWarnings("unchecked")
            E x = (E) items[takeIndex];
            items[takeIndex] = null;
            if (++takeIndex == items.length)
                takeIndex = 0;
            count--;
            if (itrs != null)
                itrs.elementDequeued();
            notFull.signal();
            return x;
        }

take的流程：

*  获取锁；
 *  判断数组的元素个数：如果等于0就在notEmpty队列中等待获取元素；
 *  如果数组有元素就获取元素
 *  从notFull队列中唤醒线程；

从notFull唤醒的线程的原因：假设数组中元素是填满的，并且此时notFull中也有线程等待将元素put到数组中；这个时候如果有线程从数组中获取到元素，数组就会空出来一个位置，而空出来的位置正好可以从notFull队列中唤醒线程，该线程可以向数组中添加元素。

put，take中有2个比较重要的变量：putIndex，takeIndex；这2个变量分别是添加元素的位置，获取元素的位置。

![在这里插入图片描述][e2ab7d43babd44de9f3dc9ea0ab0e111.png_pic_center]  
  takeIndex,putIndex分别是获取元素的位置，添加元素的位置；都是从0 -> length-1。也就是添加元素和获取元素都是从下标为0的位置开始，按下标顺序添加和获取元素的。当putIndex=length-1之后，下一个添加元素的位置就是 0；takeIndex也一样，当takeIndex=length-1，获取元素之后下一个获取元素的位置就是 0。

![在这里插入图片描述][e956688e9583419f8df05ee409701bdc.png_pic_center]

这2副图就是ArrayBlockingQueue的元素添加和获取的过程，还是挺简单的。

下面主要分析AQS中await，signal方法；

## **ConditionObject的wait方法** ##

源码：

public final void await() throws InterruptedException {
        
            //如果有中断标记，就抛异常
                if (Thread.interrupted())
                    throw new InterruptedException();
                    //将线程包装成node，加入到条件队列
                Node node = addConditionWaiter();
                //释放锁
                int savedState = fullyRelease(node);
                int interruptMode = 0;
                //不在同步队列就阻塞线程
                //当其他线程调用signal方法时，可以将node节点从条件队列转移到同步队列
                while (!isOnSyncQueue(node)) {
        
                    LockSupport.park(this);
                    if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                        break;
                }
                //获取锁：acquireQueued在上一篇文章中分析过，这里不再分析
                if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                    interruptMode = REINTERRUPT;
                if (node.nextWaiter != null) // clean up if cancelled
                    unlinkCancelledWaiters();
                if (interruptMode != 0)
                    reportInterruptAfterWait(interruptMode);
            }

首先将线程添加到条件队列中，并且清除队列中状态不为CONDITION的节点；然后释放锁，为什么需要释放该线程持有的锁呢？

在前面已经分析过不管是put添加元素，还是take获取元素为了线程安全都要先获取锁；因为是多线程环境，所以可能会有多个线程在同一时刻同时进行put，take操作。现在假设有3个put线程，3个take线程；这6个线程同时获取锁，只能有一个线程获取到锁，而其余线程都因获取不到锁而需要进入同步队列等待获取锁。而此时ArrayBlockingQueue的数组元素个数已经是最大值不能往数组中put元素了，因此获得锁的线程，会进入到条件队列中等待数组有空出的位置，才能put值；

![在这里插入图片描述][56ddf9ac07a44567834c8c00e495d9e8.png_pic_center]

如果此时不释放thread-1持有的锁，那么会造成ArrayBlockingQueue不可用了。因为此时，既不能往ArrayBlockingQueue中添加元素又不能从ArrayBlockingQueue中取出元素。任何线程对ArrayBlockingQueue的take，put操作都会被阻塞。而其余的offer，add，poll，peek等操作也会被阻塞因为获取不到锁而被阻塞在同步队列中。  
  因此，在线程进入到await方法时，需要释放该线程持有的锁。这样其他线程才有能继续对ArrayBlockingQueue进行put，take的操作，才能始终保持ArrayBlockingQueue处于可用状态。

以上图为例，当thread-1释放了锁之后，处于同步队列的其他线程会依次获取到锁；由于thread-2，thread-3都是put线程，因此获取到锁之后，依旧不能往ArrayBlockingQueue中添加元素，而是会进入到条件队列中等待被唤醒；

![在这里插入图片描述][ca57dc5d622149259203c4feae3f44a5.png_pic_center]

当thread-4获取到锁之后会从ArrayBlockingQueue获取元素，并且从notFull条件队列中唤醒第一个节点：thread-1，并将thread-1添加到同步队列的尾部；

![在这里插入图片描述][1e25ae329edf4d2e976c032a013f87e4.png_pic_center]

在thread-5，thread-6分别获取锁获取到元素之后，会分调用signal方法转移一个notFull条件队列中的一个节点，将notFull条件队列中的节点转移到同步队列中；

![在这里插入图片描述][6bf7b56f20bc4fd886a919d55d601daf.png_pic_center]  
  当thread-1，thread-2，thread-3，分别获取到锁之后，往数组添加元素时都会判断是否有空位，如果有空位置才会往数组添加元素，如果没有空位置那么还是会进入到notFull条件队列中等待被唤醒。如果有空位置，就会往数组中添加元素。

![在这里插入图片描述][f45d059aa7ea499b840d1b5d18dd2d31.png_pic_center]

## **ConditionObject的signal方法** ##

signal方法的作用就是将处于条件队列的节点转移到同步队列。这一步操作主要是在transferForSignal方法中完成

final boolean transferForSignal(Node node) {
        
            /*
             *修改节点状态：CONDITION -->  0
             */
            if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
                return false;
    
            /*
             * 将节点添加到同步队列的末尾。
             */
            Node p = enq(node);
            int ws = p.waitStatus;
            //prev节点状态异常，或者修改prev的状态失败，会将线程唤醒；
            if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
                LockSupport.unpark(node.thread);//唤醒节点
            return true;
        }

## end ##

本文主要是要搞清楚await方法中为什么要释放锁，以及condition条件队列在ArrayBlockingQueue中的应用。因此await，signal中其他比较细节的地方就不再分析了，这些细节也是非常重要的，不过这些细节不影响对condition的await，signal整体流程的认识。感兴趣的话，可以自己研究一下。

最开始想要了解AQS的condition条件队列还有一部分原因是想要了解CyclicBarrier,他与CountdownLatch的作用差不多但是实现方式不同。CountdownLatch使用的是AQS的共享锁来实现countdown，await方法。CyclicBarrier使用了条件队列来实现await方法，在还有线程没有完成任务的时候将线程阻塞到条件队列中，由最后一个线程将被阻塞的线程唤醒。

对ArrayBlockingQueue的put，take过程中在什么时候线程应该加入到notFull，notEmpty的情况搞清楚之后，还可以再看LinkedBlockingDeque，LinkedBlockingQueue他们的put，take过程中入队的条件和ArrayBlockingQueue是一样的 。这2个类的底层是链表结构：一个双向链表，一个单链表结构非常类似，构造方法中可以指定一个容量值默认是int最大值。LinkedBlockingQueue稍微有点不同，它是使用了2把锁分别管理notEmpty，notFull队列，不过原理是一样的。（可以思考一下，为什么在有LinkedBlockingQueue的情况下还需要LinkedBlockingDeque？LinkedBlockingDeque实现的功能有哪些是LinkedBlockingQueue不能实现的？为什么不能实现？）

PriorityBlockingQueue的原理和ArrayBlockingQueue也差不多，PriorityBlockingQueue底层也是数组结构，不过它是可以在容量不够的时候自增长的，在int值范围内可以一直扩容。因此PriorityBlockingQueue没有notFull队列了，因为容量不够随时增加不必再使用队列排队等空位置。并且在获取元素时只有poll（将元素从数组中取出），peek（返回元素）两种方法；当数组没有元素时直接返回null，没有使用notEmpty队列来阻塞线程。

PriorityBlockingQueue是一个优先队列，它与ArrayBlockingQueue不同的地方还在于ArrayBlockingQueue是先进先出，而PriorityBlockingQueue则不同。添加进PriorityBlockingQueue中的元素不会按添加元素的先后顺序排放。在逻辑上，他会把数组看作完全二叉堆的结构，并且会将其构建成大顶堆/小顶堆。每次取数据的时候只需要取出堆顶的数据，然后重新调整堆的结构。如果熟悉堆排序的话就看起来比较简单，这是堆排序的一部分。它构建大顶堆/小顶堆的过程与一般的还有所不同，因为一般情况下构建大/小顶堆是给你一堆乱序数据然后将乱序数字构建成大/小顶堆。而在PriorityBlockingQueue中，从第一个数据开始就按照大顶堆/小顶堆性质来构建。最开始看源码的时候还有点疑惑，就是因为没有考虑到这个原因。

juc下面还有其他同步队列，最简单的就是上面提到的这4个，搞清楚一个剩余3个就都懂了；其余的可能都要比这4个难懂，需要时间慢慢看。

*  LinkedTransferQueue：开篇就是几百行的注释，看起来很难。put，take方法也不是熟悉流程。代码很多，可能会很难看懂。查了一下，据说性能比LinkedBlockingQueue，LinkedBlockingDeque好很多，有时间再看吧。
 *  SynchronousQueue，简单看了下感觉有点复杂。
 *  DelayQueue延时队列，这个比较简单而且也比较常用。它主要是用到了优先队列PriorityQueue来实现。定时任务可以用这个来实现。

后面想到其他的再补充吧，也可能会单独分析后面提到的这3个类。对juc了解的越多就越感觉Doug Lea厉害的离谱，很难想象他一个人是怎么做到的。一个ConcurrentHashMap已经足够复杂，并且还是性能很好的高并发下的存取，查询元素的工具。又以AQS为基础，构建出了很多多线程下使用的工具。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5na2FpdGk_shadow_0_text_RG91ZyBsZWEgWVlEUw_g_south_pic_center]

[bc7219cda3e342009ddf3366d3fd1ae9.png_pic_center]: https://img-blog.csdnimg.cn/bc7219cda3e342009ddf3366d3fd1ae9.png#pic_center
[3faf33008b4c4663bd23fdc005395b52.png_pic_center]: https://img-blog.csdnimg.cn/3faf33008b4c4663bd23fdc005395b52.png#pic_center
[891fa33acb174651a179c8cdf5143742.png_pic_center]: https://img-blog.csdnimg.cn/891fa33acb174651a179c8cdf5143742.png#pic_center
[e2ab7d43babd44de9f3dc9ea0ab0e111.png_pic_center]: https://img-blog.csdnimg.cn/e2ab7d43babd44de9f3dc9ea0ab0e111.png#pic_center
[e956688e9583419f8df05ee409701bdc.png_pic_center]: https://img-blog.csdnimg.cn/e956688e9583419f8df05ee409701bdc.png#pic_center
[56ddf9ac07a44567834c8c00e495d9e8.png_pic_center]: https://img-blog.csdnimg.cn/56ddf9ac07a44567834c8c00e495d9e8.png#pic_center
[ca57dc5d622149259203c4feae3f44a5.png_pic_center]: https://img-blog.csdnimg.cn/ca57dc5d622149259203c4feae3f44a5.png#pic_center
[1e25ae329edf4d2e976c032a013f87e4.png_pic_center]: https://img-blog.csdnimg.cn/1e25ae329edf4d2e976c032a013f87e4.png#pic_center
[6bf7b56f20bc4fd886a919d55d601daf.png_pic_center]: https://img-blog.csdnimg.cn/6bf7b56f20bc4fd886a919d55d601daf.png#pic_center
[f45d059aa7ea499b840d1b5d18dd2d31.png_pic_center]: https://img-blog.csdnimg.cn/f45d059aa7ea499b840d1b5d18dd2d31.png#pic_center
[watermark_type_ZmFuZ3poZW5na2FpdGk_shadow_0_text_RG91ZyBsZWEgWVlEUw_g_south_pic_center]: https://img-blog.csdnimg.cn/d39daef97d174652a4bd51f843703754.jpeg?x-oss-process=image/watermark,type_ZmFuZ3poZW5na2FpdGk=,shadow_0,text_RG91ZyBsZWEgWVlEUw==,g_south#pic_center