Java 并发编程（十）：显示锁 Condition 接口

Myth丶恋晨 2022-04-10 09:25 195阅读 0赞

## 1、概述 ##

Condition 接口也提供了类似 Object 的监视器方法，它与 Lock 配合可以实现 等待/通知 模式，

当要实现一个等待/通知模式时，我们首先想到的就是 synchronized 同步关键字，它配合 Object 的 wait()、notify() 等系列方法实现。

还有一种实现方式，那就是使用显式锁 Lock ，配合 Condition 也可以实现等待/通知模式。

Object的监控器方法 和 Condition 接口，两者在使用方式及功能特性上还是有差别的，对比如下：

<table> 
 <thead> 
  <tr> 
   <th>对比项</th> 
   <th>Object 监视器</th> 
   <th>Condition</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>前置条件</td> 
   <td>获取对象锁</td> 
   <td>Lock.lock() 获取锁,Lock.newCondition() 获取 Condition 对象</td> 
  </tr> 
  <tr> 
   <td>调用方式</td> 
   <td>直接调用，object.wait()</td> 
   <td>直接调用,condition.await()</td> 
  </tr> 
  <tr> 
   <td>等待队列数量</td> 
   <td>1个</td> 
   <td>多个</td> 
  </tr> 
  <tr> 
   <td>当前线程释放锁并进入等待状态</td> 
   <td>支持</td> 
   <td>支持</td> 
  </tr> 
  <tr> 
   <td>当前线程释放锁并进入等待状态，在等待状态中不响应中断</td> 
   <td>不支持</td> 
   <td>支持</td> 
  </tr> 
  <tr> 
   <td>当前线程释放锁并进入超时等待状态</td> 
   <td>支持</td> 
   <td>支持</td> 
  </tr> 
  <tr> 
   <td>当前线程释放锁并进入等待状态直到将来某个时间</td> 
   <td>不支持</td> 
   <td>支持</td> 
  </tr> 
  <tr> 
   <td>唤醒等待队列中的一个线程</td> 
   <td>支持</td> 
   <td>支持</td> 
  </tr> 
  <tr> 
   <td>唤醒等待队列中的所有线程</td> 
   <td>支持</td> 
   <td>支持</td> 
  </tr> 
 </tbody> 
</table>

## 2、Condition 实现分析 ##

Condition 定义了 等待/通知 两种类型的方法，当前线程调用这些方式之前，必须获取 Condition 对象相关联的锁。

Condition 是一个接口，  
Condition 的实现类是 Lock（AQS）中的 ConditionObject。  
Lock 接口中有个 newCondition() 方法，通过这个方法可以获得 Condition 对象（就是 ConditionObject）。

Condition 的使用方式比较简单，需要注意的是在调用方法前必须先获取锁：

Lock lock  = new ReentrantLock();
    Condition c1 = lock.newCondition();
    Condition c2 = lock.newCondition();

lock.newCondition() 返回的是Condition的一个实现，该类在AbstractQueuedSynchronizer中被实现，该实现类为 ConditionObject

public class ConditionObject implements Condition, java.io.Serializable {
            private transient Node firstWaiter;
            private transient Node lastWaiter;
            ...

可以看到，等待队列和同步队列一样，使用的都是同步器 AQS 中的节点类 Node。  
同样拥有首节点和尾节点，每个 Condition 对象都包含着一个 FIFO 队列。

### 2.1、Condition 接口方法说明 ###

Condition 接口包含以下方法：

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2NDcwMzU_size_16_color_FFFFFF_t_70]

#### 2.1.1、void await() throws InterruptedException： ####

当前线程进入等待状态直到被通知（signal）或者中断；

当前线程进入等待状态，直到被唤醒的场景有以下情况：

（1）其他线程调用相同 Condition 对象的 signal/signalAll 方法，当前线程被唤醒；

（2）其他线程调用 interrupt 方法中断了当前线程；

#### 2.1.2、void awaitUninterruptibly() ####

当前线程进入等待状态直到被通知，在此过程中对中断信号不敏感，不支持中断当前线程，直到其它线程调用了相同 Condition 对象的 signal/signalAll 方法，当前线程才能被唤醒

#### 2.1.3、long awaitNanos(long nanosTimeout) throws InterruptedException; ####

当前线程进入等待状态，直到被通知（signal/signalAll）、中断（interrupt）或者超时才会被唤醒

#### 2.1.4、boolean await(long time, TimeUnit unit) throws InterruptedException; ####

当前线程进入等待状态，直到被通知（signal/signalAll）、中断（interrupt）或者过了指定时间才会被唤醒

#### 2.1.5、boolean awaitUntil(Date deadline) throws InterruptedException; ####

当前线程进入等待状态，直到被通知、中断或者超时。如果没到指定时间被通知，则返回true，否则返回false

#### 2.1.6、void signal(); ####

唤醒一个等待在Condition上的线程，被唤醒的线程在方法返回前必须获得与Condition对象关联的锁

#### 2.1.7、void signalAll(); ####

唤醒所有等待在Condition上的线程，能够从await()等方法返回的线程必须先获得与Condition对象关联的锁

### 2.2、等待 await() 方法实现分析 ###

public final void await() throws InterruptedException {
                if (Thread.interrupted())
                    throw new InterruptedException();
                Node node = addConditionWaiter();
                //释放同步状态（锁）
                long savedState = fullyRelease(node);
                int interruptMode = 0;
                //判断节点是否放入同步对列
                while (!isOnSyncQueue(node)) {
                    //阻塞
                    LockSupport.park(this);
                    //如果已经中断了，则退出
                    if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                        break;
                }
                if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                    interruptMode = REINTERRUPT;
                if (node.nextWaiter != null) // clean up if cancelled
                    unlinkCancelledWaiters();
                if (interruptMode != 0)
                    reportInterruptAfterWait(interruptMode);
            }

分析上述方法的大概过程：

1、将当前线程创建为节点，加入等待队列；

2、释放锁，唤醒同步队列中的后继节点；

3、while循环判断节点是否放入同步队列：

*  没有放入，则阻塞，继续 while 循环（如果已经中断了，则退出）
 *  已经放入，则退出 while 循环，执行后面的判断

4、退出 while 说明节点已经在同步队列中，调用 acquireQueued() 方法加入同步状态竞争。

5、竞争到锁后从 await() 方法返回，即退出该方法。

#### 2.2.1、addConditionWaiter() 方法： ####

该方法是添加新的节点到等待队列并返回该节点

private Node addConditionWaiter() {
                Node t = lastWaiter;
                //清除条件队列中所有状态不为 CONDITION 的节点
                if (t != null && t.waitStatus != Node.CONDITION) {
                    unlinkCancelledWaiters();
                    t = lastWaiter;
                }
                Node node = new Node(Thread.currentThread(), Node.CONDITION);
                if (t == null)
                    firstWaiter = node;
                else
                    t.nextWaiter = node;
                lastWaiter = node;
                return node;
            }

同步队列的首节点移动到等待队列。加入尾节点之前会清除所有状态不为 Condition 的节点。

### 2.3、通知 signal() 方法实现分析 ###

调用 Condition 的 signal() 方法，可以唤醒等待队列的首节点（等待时间最长），唤醒之前会将该节点移动到同步队列中。

public final void signal() {
                //如果同步状态不是被当前线程独占，直接抛出异常。
                if (!isHeldExclusively())
                    throw new IllegalMonitorStateException();
                Node first = firstWaiter;
                
                //通知等待队列队首的节点
                if (first != null)
                    doSignal(first);
            }
            
            
            private void doSignal(Node first) {
                do {
                    if ( (firstWaiter = first.nextWaiter) == null)
                        lastWaiter = null;
                    first.nextWaiter = null;
                } while (!transferForSignal(first) &&
                         (first = firstWaiter) != null);
            }
            
            
            final boolean transferForSignal(Node node) {
            //将节点状态变为0 
            if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
                return false;
    
            //将该节点加入同步队列
            Node p = enq(node);
            int ws = p.waitStatus;
            //如果结点p的状态为cancel 或者修改waitStatus失败，则直接唤醒
            if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
                LockSupport.unpark(node.thread);
            return true;
        }

1、先判断当前线程是否获取了独占锁；

2、然后对首节点调用 doSignal() 方法，修改首节点

3、调用 transferForSignal() 方法将节点移动到同步队列

4、调用同步器的 enq 方法，将节点移动到同步队列

5、满足条件后使用 LockSupport 唤醒该线程

**signalAll() 方法**

将等待队列中的全部节点移动到同步队列中，并唤醒每个节点的线程

public final void signalAll() {
                if (!isHeldExclusively())
                    throw new IllegalMonitorStateException();
                Node first = firstWaiter;
                if (first != null)
                    doSignalAll(first);
            }
            
            private void doSignalAll(Node first) {
                lastWaiter = firstWaiter = null;
                do {
                    Node next = first.nextWaiter;
                    first.nextWaiter = null;
                    transferForSignal(first);
                    first = next;
                } while (first != null);
            }

doSignalAll() 方法使用了 do-while 循环来唤醒每一个等待队列中的节点，直到 first 为 null 时，停止循环。

### 2.4、等待/通知 过程总结 ###

1、一个线程获取锁后，通过调用 Condition 的 await() 方法，会将当前线程先加入到等待队列中，并释放锁。然后就在 await() 中的一个 while 循环中判断节点是否已经在同步队列，是则尝试获取锁，否则一直阻塞。

2、当线程调用 signal() 方法后，程序首先检查当前线程是否获取了锁，然后通过 doSignal(Node first) 方法将节点移动到同步队列，并唤醒节点中的线程。

3、被唤醒的线程，将从 await() 中的 while 循环中退出来，然后调用 acquireQueued() 方法竞争同步状态。竞争成功则退出 await() 方法，继续执行。

## 3、Condition 使用示例 ##

下面通过一个有界队列的示例来了解 Condition 的使用方式。有界队列是一种特殊队列，当队列为空时，队列的获取操作将会阻塞获取的线程，直到队列中有新增加的元素；当队列满时，队列的插入操作将会阻塞插入的线程，直到队列有了可用空间。

package com.lkf.condition;
    
    import java.util.Arrays;
    import java.util.concurrent.locks.Condition;
    import java.util.concurrent.locks.Lock;
    import java.util.concurrent.locks.ReentrantLock;
    
    /**
     * 有界队列
     *
     * @author kaifeng
     * @date 2019/1/1
     */
    public class BoundedQueue {
        //定义为数组，在创建对象时就确定容量
        private Integer[] items;
        private Lock lock = new ReentrantLock();
        private Condition notEmpty = lock.newCondition();
        private Condition notFull = lock.newCondition();
    
        // 添加的下标，删除的下标和数组当前元素数量
        private int addIndex, removeIndex, count;
    
        public BoundedQueue(int size) {
            items = new Integer[size];
        }
    
        /**
         * 添加一个元素，如果数组满了，则添加线程进入等待状态
         *
         * @param object 添加的元素
         */
        public void add(Integer object) throws InterruptedException {
            lock.lock();
            try {
                while (count == items.length) {
                    notFull.await();
                }
                items[addIndex] = object;
                if (++addIndex == items.length) {
                    addIndex = 0;
                }
                count++;
                System.out.println(Thread.currentThread() + " 新增一个元素，当前数组为：" + Arrays.toString(items));
                notEmpty.signal();
            } finally {
                lock.unlock();
            }
        }
    
        /**
         * 从头部删除一个元素，如果数组为空，则删除线程进入等待状态，直到有新元素添加
         */
        @SuppressWarnings("unchecked")
        public Integer remove() throws InterruptedException {
            lock.lock();
            try {
                while (count == 0) {
                    notEmpty.await();
                }
                Integer temp = items[removeIndex];
                items[removeIndex] = null;
                System.out.println(Thread.currentThread() + " 读取一个元素，当前数组为：" + Arrays.toString(items));
                if (++removeIndex == items.length) {
                    removeIndex = 0;
                }
                count--;
                notFull.signal();
                return temp;
            } finally {
                lock.unlock();
            }
    
    
        }
    }

有界队列测试客户端：

package com.lkf.condition;
    
    import java.util.Random;
    import java.util.function.Consumer;
    
    /**
     * 有界队列测试
     *
     * @author kaifeng
     * @date 2019/1/1
     */
    public class ConditionDemo {
        private final static Random random = new Random();
    
        public static void main(String[] args) {
            BoundedQueue queue = new BoundedQueue(5);
            for (int i = 0; i < 5; i++) {
                Thread thread = new Thread(new ProducterRunable(queue), "ProdducterThread_".concat(String.valueOf(i)));
                thread.start();
            }
    
            for (int i = 0; i < 5; i++) {
                Thread thread = new Thread(new ConsumerRunable(queue), "ConsumerThread_".concat(String.valueOf(i)));
                thread.start();
            }
    
        }
    
        /**
         * 插入元素到队列
         */
        static class ProducterRunable implements Runnable {
            private BoundedQueue queue;
    
            public ProducterRunable(BoundedQueue queue) {
                this.queue = queue;
            }
    
            public void produce() throws InterruptedException {
                queue.add(new Integer(random.nextInt(100)));
            }
    
            @Override
            public void run() {
                try {
                    produce();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    
        /**
         * 从队列读取元素
         */
        static class ConsumerRunable implements Runnable {
            private BoundedQueue queue;
    
            public ConsumerRunable(BoundedQueue queue) {
                this.queue = queue;
            }
    
            public Integer remove() throws InterruptedException {
                return queue.remove();
            }
    
            @Override
            public void run() {
                try {
                    remove();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }
    
    }

测试结果如下：

Thread[ProdducterThread_0,5,main] 新增一个元素，当前数组为：[91, null, null, null, null]
    Thread[ProdducterThread_2,5,main] 新增一个元素，当前数组为：[91, 79, null, null, null]
    Thread[ProdducterThread_1,5,main] 新增一个元素，当前数组为：[91, 79, 24, null, null]
    Thread[ProdducterThread_3,5,main] 新增一个元素，当前数组为：[91, 79, 24, 30, null]
    Thread[ProdducterThread_4,5,main] 新增一个元素，当前数组为：[91, 79, 24, 30, 70]
    Thread[ConsumerThread_0,5,main] 读取一个元素，当前数组为：[null, 79, 24, 30, 70]
    Thread[ConsumerThread_1,5,main] 读取一个元素，当前数组为：[null, null, 24, 30, 70]
    Thread[ConsumerThread_2,5,main] 读取一个元素，当前数组为：[null, null, null, 30, 70]
    Thread[ConsumerThread_3,5,main] 读取一个元素，当前数组为：[null, null, null, null, 70]
    Thread[ConsumerThread_4,5,main] 读取一个元素，当前数组为：[null, null, null, null, null]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3UwMTA2NDcwMzU_size_16_color_FFFFFF_t_70]: /images/20220401/4df00df04bf84379a4e6b0b6cb9e950b.png