八：深入理解 Condition —— 线程通信（等待/通知）

今天药忘吃喽~ 2024-05-05 07:44 71阅读 0赞

#### 目录 ####

*  1、背景
 *  2、Condition 入门
 *  3、Condition 源码解析
 *   *  3.1、\`Lock\` 与 \`Condition\` 的关系
     *  3.2、\`await()\` 方法 —— AQS.ConditionObject
     *   *  3.2.1、\`addConditionWaiter()\` 方法 —— AQS.ConditionObject
         *   *  3.2.1.1、\`unlinkCancelledWaiters()\` 方法 —— AQS.ConditionObject
         *  3.2.2、\`fullyRelease()\` 方法 —— AQS
         *   *  3.2.2.1 \`release()\` 方法 —— AQS
         *  3.2.3、\`isOnSyncQueue()\` 方法 —— AQS
         *   *  3.2.3.1、\`findNodeFromTail()\` 方法 —— AQS
         *  3.2.4、\`checkInterruptWhileWaiting()\` 方法 —— AQS.ConditionObject
         *   *  3.2.4.1、\`transferAfterCancelledWait()\` 方法
         *  3.2.5、\`reportInterruptAfterWait()\` 方法 —— AQS.ConditionObject
     *  3.3、\`signal()\` 方法 —— AQS.ConditionObject
     *   *  3.3.1、\`doSignal()\` 方法
         *   *  3.3.1.1 \`transferForSignal()\` 方法
     *  3.4、\`signalAll()\` 方法 —— AQS.ConditionObject
 *  4、总结

## 1、背景 ##

任意一个Java对象，都拥有一组监视器方法（定义在Object类中），主要包括 `wait()`，`notify()`，`notifyAll()` 方法，这些方法与 `synchornized` 关键字相配合，可以实现**等待/通知**模式【至少有两个线程】

案例：

> 有2个字符串，一个字符串是1234567，另一个字符串是ABCDEFG，然后需要按1A2B3C4D5E6F7G或者A1B2C3D45E6F7G这种输出【两个线程交替执行】

public class Test {
        
    
        public static void main(String[] args) {
        
            final char arrayA[] = "1234567".toCharArray();
            final char arrayB[] = "ABCDEFG".toCharArray();
    
            Object lock = new Object();
    
            Runnable taskA = () -> {
        
                synchronized (lock) {
        
                    for (char ch : arrayA) {
        
                        System.out.print(ch);
                        lock.notify();
                        try {
        
                            lock.wait();
                        } catch (InterruptedException e) {
        
                            throw new RuntimeException(e);
                        }
                    }
                    lock.notify();
                }
            };
            Runnable taskB = () -> {
        
                synchronized (lock) {
        
                    for (char ch : arrayB) {
        
                        System.out.print(ch);
                        lock.notify();
                        try {
        
                            lock.wait();
                        } catch (InterruptedException e) {
        
                            throw new RuntimeException(e);
                        }
                    }
                    lock.notify();
                }
            };
            new Thread(taskA, "线程A").start();
            new Thread(taskB, "线程B").start();
        }
    
    }

[两个线程交替执行的几种方式][Link 1]  
[Java实现多线程循环输出abc][Java_abc]

*  `wait()`：等待。释放对象锁，阻塞当前线程，并将自己加入到条件等待队列中去
 *  `notify()`：唤醒某一个线程。将条件等待队列中的线程转移到同步等待队列中，待锁释放后，进行竞争锁
 *  `notifyAll()`：唤醒所有等待的线程

## 2、Condition 入门 ##

Condition 接口也提供了类似的 Object 的监视器方法，与 Lock 配合可以实现等待/通知模式【Condition 是依赖 Lock 对象的】。但是这两者在使用方式以及功能上还是有差别的

![在这里插入图片描述][e042f56cd374461b842ba0dcfcbe65f9.png]

> 案例：实现一个非常典型的生产者和消费者模式

生产者：

public class Producer implements Runnable{
        
    
        private Queue<String> msg;
        private int maxSize;
        private Lock lock;
        private Condition condition;
    
        public Producer(Queue<String> msg, int maxSize, Lock lock, Condition condition) {
        
            this.msg = msg;
            this.maxSize = maxSize;
            this.lock = lock;
            this.condition = condition;
        }
    
        @Override
        public void run() {
        
            int i = 0;
            while (true) {
        
                i++;
                lock.lock();
                try {
        
                    while (msg.size() == maxSize) {
        
                        // 队列中消息满了，此时生产者不能再生产了，因为装不下了，所以生产者就开始等待状态
                        System.out.println("生产者队列满了，先等待");
                        // 阻塞线程并释放锁
                        condition.await();
                    }
                    // 生产消息
                    Thread.sleep(1000);
                    System.out.println("生产消息：" + i);
                    msg.add("生产者的消息内容" + i);
                    // 唤醒阻塞状态下的线程
                    condition.signal();
                } catch (Exception e) {
        
                    e.printStackTrace();
                } finally{
        
                    lock.unlock();
                }
            }
        }
    
    }

消费者：

public class Consumer implements Runnable {
        
    
        private Queue<String> msg;
        private int maxSize;
        private Lock lock;
        private Condition condition;
    
        public Consumer(Queue<String> msg, int maxSize, Lock lock, Condition condition) {
        
            this.msg = msg;
            this.maxSize = maxSize;
            this.lock = lock;
            this.condition = condition;
        }
    
        @Override
        public void run() {
        
            int i = 0;
            while (true) {
        
                i++;
                lock.lock();
                try {
        
                    // 消费者进来的时候需要判断是有可用的消息，没有可用的消息就等待状态
                    while (msg.isEmpty()) {
        
                        System.out.println("消费者队列空了，先等待");
                        // 阻塞线程并释放锁
                        condition.await();
                    }
                    // 消费消息
                    Thread.sleep(1000);
                    System.out.println("消费消息：" + msg.remove());
                    // 唤醒阻塞状态下的线程
                    condition.signal();
                } catch (Exception e) {
        
                    e.printStackTrace();
                } finally{
        
                    lock.unlock();
                }
            }
        }
    
    }

测试：

public static void main(String[] args){
        
        Queue<String> queue = new LinkedList<>();
        Lock lock = new ReentrantLock();
        // 返回是一个 ConditionObject 对象
        Condition condition = lock.newCondition();
        int maxSize = 5;
        Producer producer = new Producer(queue, maxSize, lock, condition);
        Consumer consumer = new Consumer(queue, maxSize, lock, condition);
    
        Thread t1 = new Thread(producer);
        Thread t2 = new Thread(consumer);
        t1.start();
        t2.start();
    }

## 3、Condition 源码解析 ##

### 3.1、`Lock` 与 `Condition` 的关系 ###

public class ReentrantLock implements Lock {
        
    	private final Sync sync;
    	// AQS 类
    	abstract static class Sync extends AbstractQueuedSynchronizer {
        
    		final ConditionObject newCondition() {
        
                return new ConditionObject();
            }
    	}
    	
    	// 通过 AQS 返回 Condition 对象
    	public Condition newCondition() {
        
            return sync.newCondition();
        }
    }

`AQS`：有一个内部类 `ConditionObject`

public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer {
        
    	
    	// AQS 中的一个内部类
    	public class ConditionObject implements Condition {
        
    		// 单向链表头节点
    		private transient Node firstWaiter;
    		// 单向链表尾节点
    		private transient Node lastWaiter;
    		
    		// 唤醒后，重新中断下
    		private static final int REINTERRUPT =  1;
    		// 唤醒后，抛异常
    		private static final int THROW_IE    = -1;
    	}
    	
    	static final class Node {
        
    		// 当前节点对应的线程
    		volatile Thread thread;
    		// 当前节点后驱
    		Node nextWaiter;
    	}
    }

`Condition` 的操作需要相关联的锁，每一个 `Condition` 对象可以阻塞多个线程，内部使用了单向链表。其中，`firstWaiter` 指向了头节点，`lastWaiter` 指向了尾节点。

条件等待队列单向链表如下：

![在这里插入图片描述][db1828e4db084ce9b02863d0e35d906b.png]  
在 Object 的监视器模型上，一个对象有一个同步等待队列、条件等待队列，而 `Lock` 中有一个同步等待队列和多个条件等待队列。如下图：

![在这里插入图片描述][1178be57d2ba4454919de6d8fdf20a73.png]

### 3.2、`await()` 方法 —— AQS.ConditionObject ###

public final void await() throws InterruptedException {
        
    	// 如果当前线程中断，则抛异常
        if (Thread.interrupted()) {
        
        	throw new InterruptedException();
       	}
       	// 将需要阻塞的当前线程添加到条件等待等待中（遍历移除链表中已取消的节点）
        Node node = addConditionWaiter();
        // 彻底释放锁，并唤醒同步队列中阻塞的一个线程
        int savedState = fullyRelease(node);
        int interruptMode = 0;
        // 判断当前节点是否在同步队列
        while (!isOnSyncQueue(node)) {
        
        	// 如果不在同步等待队列中，则阻塞自己【因为当前节点没有被唤醒去争抢同步锁，直到其他的线程调用 signal() 唤醒】
            LockSupport.park(this);
            // 线程唤醒有两种可能：signal()/signalAll()【unpark() 方法】；被中断
            // 判断线程在阻塞过程中的中断模式
            if ((interruptMode = checkInterruptWhileWaiting(node)) != 0) {
        
            	// 线程中断，退出 while() 循环
            	break;
           	}
        }
        // 1：被调用了 signal() 方法，在同步队列中
        // 2：被中断【REINTERRUPT：调用了 signal() 方法，会在同步队列中；THROW_IE：未调用signal() 方法，没有在同步队列中】
        // 尝试获取锁：被调用了 signal() 方法 | REINTERRUPT
        if (acquireQueued(node, savedState) && interruptMode != THROW_IE) {
        
        	// 如果获取锁成功，且线程发生中断，那需要进行中断
        	interruptMode = REINTERRUPT;
       	}
       	// THROW_IE：清理条件队列中的 CANCEL 状态的节点
        if (node.nextWaiter != null) {
        
        	unlinkCancelledWaiters();
       	}
       	// 进行中断
        if (interruptMode != 0) {
        
        	reportInterruptAfterWait(interruptMode);
       	}   
    }

`await()` 方法：使当前线程【已持有锁】进入条件等待队列并释放锁【等待其它线程调用 `signal()/signalAll()` 方法】

#### 3.2.1、`addConditionWaiter()` 方法 —— AQS.ConditionObject ####

private Node addConditionWaiter() {
        
        Node t = lastWaiter;
        if (t != null && t.waitStatus != Node.CONDITION) {
        
        	// 如果存在尾节点，且其的状态是取消状态，则循环遍历链表移除“取消状态”的节点
            unlinkCancelledWaiters();
            t = lastWaiter;
        }
        // new 一个节点，状态为 CONDITION，并添加到链表中
        Node node = new Node(Thread.currentThread(), Node.CONDITION);
        if (t == null) {
        
            firstWaiter = node;    
        } else {
        
            t.nextWaiter = node;    
        }
        lastWaiter = node;
        return node;
    }

`addConditionWaiter()` 方法：如果条件等待队列上存在尾节点，且其状态为【CANCEL】，则从头至尾遍历条件等待队列，去掉状态为【CANCEL】的节点，方便 GC【避免没有调用 `signal()/signalAll()` 方法造成了垃圾滞留】；再 new 一个【CONDITION】的节点，放入队尾

> ①：对于 Condition 节点，它的状态为：CONDITION【默认】、CANCEL  
> ②：在调用 await() 方法时，已经获取了锁，所以，这里节点入队时并未使用 CAS 操作，线程安全

##### 3.2.1.1、`unlinkCancelledWaiters()` 方法 —— AQS.ConditionObject #####

private void unlinkCancelledWaiters() {
        
    	// 当前移动的节点
        Node t = firstWaiter;
        // trail：永远是条件等待队列中最后一个为【CONDITION】 状态的节点
        Node trail = null;
        while (t != null) {
        
        	// 如果条件等待队列不为空，则从头至尾循环遍历
            Node next = t.nextWaiter;
            if (t.waitStatus != Node.CONDITION) {
        
            	// 处理节点状态为【CANCEL】
                t.nextWaiter = null;
                if (trail == null) {
        
                    // 前 N（N >= 1）个节点的状态都是 CANCEL 状态
                    firstWaiter = next;            
                } else {
        
                	// 去掉当前节点 t
                    trail.nextWaiter = next;            
                }
                if (next == null) {
        
                    // 处理尾节点
                    lastWaiter = trail;            
                }
            } else {
        
                // 如果当前节点为 CONDITION状态，则移动 trail 到当前节点
                trail = t;
            }
            // 往后移一个节点
            // 当前节点为 CONDITION 状态，则往后移
            // 当前节点为 CANCEL 状态，则去掉当前节点：trail.nextWaiter = next;  
            t = next;
        }
    }

`unlinkCancelledWaiters()` 方法：如果条件等待队列不为空，则从头至尾循环遍历，去掉 CANCEL 状态的节点，方便 GC

#### 3.2.2、`fullyRelease()` 方法 —— AQS ####

final int fullyRelease(Node node) {
        
        boolean failed = true;
        try {
        
            // 1.获取 state：state >= 1【可重入】
            int savedState = getState();
            // 2.释放锁并唤醒下一个同步队列中的线程（彻底释放：可重入锁。如果 state = 2，也要一次性释放掉，通过调用 release(state) 方法）
            if (release(savedState)) {
        
                failed = false;
                return savedState;
            } else {
        
                throw new IllegalMonitorStateException();
            }
        } finally {
        
            if (failed) {
        
                node.waitStatus = Node.CANCELLED;        
            }
        }
    }

`fullyRelease()` 方法：彻底释放同步锁，如果成功，则唤醒同步队列中下一个的线程

##### 3.2.2.1 `release()` 方法 —— AQS #####

public final boolean release(int arg) {
        
        // 释放锁
        if (tryRelease(arg)) {
        
            // 如果释放锁成功，则唤醒同步队列中下一个的线程
            Node h = head;
            if (h != null && h.waitStatus != 0) {
        
                // 唤醒同步队列中下一个的线程
                unparkSuccessor(h);
            }
            return true;
        }
        return false;
    }

#### 3.2.3、`isOnSyncQueue()` 方法 —— AQS ####

final boolean isOnSyncQueue(Node node) {
        
        // 如果当前节点状态是 CONDITION 或 node.prev 是 null，则证明当前节点在条件队列上而不是同步队列上。
        // 之所以可以用 node.prev 来判断，是因为一个节点如果要加入同步队列，在加入前就会设置好 prev 字段
        if (node.waitStatus == Node.CONDITION || node.prev == null) {
        
            return false;    
        }
        // 如果 node.next 不为 null，则一定在同步队列上，因为 node.next 是在节点加入同步队列后设置的
        if (node.next != null) {
        
            return true;    
        }
        // 前面的两个判断没有返回的话，就从同步队列队尾遍历一个一个看是否包含当前节点
        return findNodeFromTail(node);
    }

`isOnSyncQueue()` 方法：判断当前节点是否在同步等待队列中

> 为什么要做这个判断呢？

因为在 Condition 队列中的节点会重新加入到 AQS 队列去竞争锁。也就是当调用 `signal()/signalAll()` 的时候，会把当前节点从 Condition 队列转移到 AQS 队列

##### 3.2.3.1、`findNodeFromTail()` 方法 —— AQS #####

private boolean findNodeFromTail(Node node) {
        
        Node t = tail;
        for (;;) {
        
            if (t == node) {
        
                return true;        
            }
            if (t == null) {
        
                return false;        
            }
            t = t.prev;
        }
    }

`findNodeFromTail()` 方法：在同步队列上从尾至头遍历，看当前节点是不是在同步等待队列上

#### 3.2.4、`checkInterruptWhileWaiting()` 方法 —— AQS.ConditionObject ####

private int checkInterruptWhileWaiting(Node node) {
        
        return Thread.interrupted() ? (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) : 0;
    }

如果线程在阻塞过程中没有中断，则返回 0；否则【线程发生过中断】，由 `transferAfterCancelledWait()` 方法决定：如果返回 true，即：在 `signal()/signalAll()` 方法调用之前中断了，则返回 `THROW_IE`；否则，返回 `REINTERRUPT`

>  *  `THROW_IE`：抛出中断异常。其它线程还没有调用 `signal()/signalAll()` 方法
>  *  `REINTERRUPT`：线程中断。其它线程已调用 `signal()/signalAll()` 方法

##### 3.2.4.1、`transferAfterCancelledWait()` 方法 #####

final boolean transferAfterCancelledWait(Node node) {
        
    	// 执行到这里，说明线程已经发生过中断。只需要判断是否调用了 signal() 方法
    	if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {
        
    		// CAS 操作修改节点状态【CONDITION => 0】成功，说明还没有调用 signal() 方法，将节点加入到同步队列中去
    		// 【signal() 方法会会修改节点状态，并将条件等待队列中的节点转移到同步队列中去，这样，CAS 操作是会失败的】，
    	    enq(node);
    	    return true;
    	}
    	// 已经执行了 signal() 方法，修改了节点状态。只需要判断是否在同步队列中。
    	// 如果不在，只需要等待 signal() 将节点加入到同步队列即可
    	while (!isOnSyncQueue(node)) {
        
    		Thread.yield();
    	}
    	return false;
    }

#### 3.2.5、`reportInterruptAfterWait()` 方法 —— AQS.ConditionObject ####

private void reportInterruptAfterWait(int interruptMode) throws InterruptedException {
        
        if (interruptMode == THROW_IE) {
        
        	throw new InterruptedException();
        } else if (interruptMode == REINTERRUPT) {
        
        	// 补上中断标记
    		selfInterrupt();
    	}
    }

### 3.3、`signal()` 方法 —— AQS.ConditionObject ###

public final void signal() {
        
        // 先判断当前线程是否获得了锁，这个判断比较简单，直接用获得锁的线程和当前线程相比即可
        if (!isHeldExclusively()) {
        
            throw new IllegalMonitorStateException();    
        }
        Node first = firstWaiter;
        if (first != null) {
        
            // 唤醒条件等待队列中节点
            doSignal(first);    
        }
    }

#### 3.3.1、`doSignal()` 方法 ####

private void doSignal(Node first) {
        
        do {
        
            // 从 Condition 队列中删除 first 节点
            if ((firstWaiter = first.nextWaiter) == null) {
        
            	// 如果没有后驱节点，则将尾节点置为空
                lastWaiter = null;        
            }
            // 从条件等待队列里移除头节点
            first.nextWaiter = null;
            // transferForSignal() 方法尝试唤醒当前节点，如果唤醒失败，则继续尝试唤醒当前节点的后继节点，如果成功，则直接退出循环
        } while (!transferForSignal(first) && (first = firstWaiter) != null);
    }

1.  删除条件等待节点的头节点
2.  将头节点添加到同步等待队列的 tail 之后，如果添加成功，则退出循环；否则，如果条件等待队列中还有节点，则后移，循环步骤 1、2
    
    1.  修改头结点的状态【CONDITION => 0】
    2.  将头节点添加到同步等待队列的 tail 之后
    3.  如果原 tail 节点状态为取消状态 | CAS 操作原 tail 节点为 SIGNAL 失败，则唤醒当前节点对应的线程【否则，基于 AQS 队列的机制来唤醒：当持有锁的线程执行完之后释放了锁，再竞争锁】

##### 3.3.1.1 `transferForSignal()` 方法 #####

final boolean transferForSignal(Node node) {
        
        // 更新节点的状态为 0，如果更新失败，只有一种可能就是节点被 CANCELLED 了
        if (!compareAndSetWaitStatus(node, Node.CONDITION, 0)) {
        
            return false;    
        }
        // 调用 enq，把条件等待队列的当前节点添加到 AQS 队列。并且返回 AQS 队列中的原 tail 节点【当前节点的前驱节点】
        Node p = enq(node);
        int ws = p.waitStatus;
        // 如果前一个节点的状态被取消了, 或者尝试设置前一个节点的状态为 SIGNAL（它的 next 节点需要唤醒）失败了
        if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL)) {
        
            // 唤醒当前节点
            LockSupport.unpark(node.thread);    
        }
        // 如果 node 的 prev 节点已经是signal 状态，那么被阻塞的 ThreadA 的唤醒工作由 AQS 队列来完成
        return true;
    }

### 3.4、`signalAll()` 方法 —— AQS.ConditionObject ###

private void doSignalAll(Node first) {
        
        lastWaiter = firstWaiter = null;
        do {
        
            // 一直循环，直至条件队列为空
            Node next = first.nextWaiter;
            first.nextWaiter = null;
            transferForSignal(first);
            first = next;
        } while (first != null);
    }

`signal()` 只是将等待队列头部节点移动到同步队列，而 `signalAll()` 将等待队列里的所有节点移动到同步队列

## 4、总结 ##

当一个持有锁的线程调用 `Condition.await()` 时，它会执行以下步骤：

1.  将当前线程加入到条件等待队列中，状态为【CONDITION】
2.  彻底释放锁，并唤醒同步队列中阻塞的一个线程
3.  判断自己是否在同步等待队列中，如果不在，将自己阻塞起来
4.  如果被 `signal()` 方法唤醒了，则去竞争同步锁

当一个持有锁的线程调用 `Condition.signal()` 时，它会执行以下操作：

[Link 1]: https://blog.csdn.net/qq_33556185/article/details/130844576
[Java_abc]: https://blog.csdn.net/qq_46130027/article/details/131655275
[e042f56cd374461b842ba0dcfcbe65f9.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/05/1f5e2a5519d04726836e6425665b7123.png
[db1828e4db084ce9b02863d0e35d906b.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/05/eeb1d44e1bd04f2b99c9da1ff1dd45f0.png
[1178be57d2ba4454919de6d8fdf20a73.png]: https://image.dandelioncloud.cn/pgy_files/images/2024/05/05/195a0493723b4efe8c7ae6c94c861b3c.png