并发编程-Condition底层设计

朱雀 2023-10-10 18:13 36阅读 0赞

`Condition` 是一个多线程协调通信的工具类，可以让某些线 程一起等待某个条件（condition），只有满足条件时，线程 才会被唤醒，它通常与`Lock`（锁）一起使用，用于在多个线程之间协调和同步访问共享资源。`Condition` 接口提供了类似于`Object.wait()`和`Object.notify()`方法的机制，但是它更加灵活，可以在等待和通知的过程中更精细地控制线程。`Condition` 接口提供了以下方法：

*  `await()`：使当前线程等待，直到其他线程调用`signal()`或`signalAll()`方法来唤醒它。
 *  `awaitUninterruptibly()`：与`await()`类似，但是它不会因为线程的中断而返回。
 *  `signal()`：唤醒一个等待在该条件上的线程，如果有多个线程等待，只唤醒其中一个线程。
 *  `signalAll()`：唤醒所有等待在该条件上的线程。
 *  `awaitNanos(long nanosTimeout)`：在指定的时间内等待，如果在超时时间之前没有被唤醒，则返回。

`Condition` 的典型用法是在一个`Lock`对象上创建多个条件，每个条件代表不同的等待队列。线程可以等待特定条件的发生，从而在特定的时刻获取锁。例如，如果多个线程需要访问一个有限资源池，可以为每个资源池创建一个条件，并让线程等待对应的条件。当一个资源可用时，可以使用对应的条件来通知等待该资源的线程，使其能够获取该资源。

#### Condition使用示例 ####

java

复制代码

`package org.example; import java.util.LinkedList; import java.util.Queue; import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.Lock; import java.util.concurrent.locks.ReentrantLock; public class Example { private final Queue<Integer> queue = new LinkedList<>(); private final Lock lock = new ReentrantLock(); private final Condition condition = lock.newCondition(); public void produce() { int i = 0; while (true) { lock.lock(); i++; // 等待队列不满 while (queue.size() == 3) { try { condition.await(); } catch (InterruptedException e) { throw new RuntimeException(e); } } // 生产一个元素 try { Thread.sleep(1000); } catch (InterruptedException e) { throw new RuntimeException(e); } queue.add(i); System.out.println("Produced: " + i); // 唤醒一个等待的消费者线程 condition.signal(); lock.unlock(); } } public void consume() { while (true) { // 等待队列不为空 lock.lock(); while (queue.isEmpty()) { try { condition.await(); } catch (InterruptedException e) { throw new RuntimeException(e); } } // 消费一个元素 try { Thread.sleep(1000); } catch (InterruptedException e) { throw new RuntimeException(e); } int num = queue.remove(); System.out.println("Consumed: " + num); // 唤醒一个等待的生产者线程 condition.signal(); lock.unlock(); } } public static void main(String[] args) { Example example = new Example(); Thread t = new Thread(example::produce); Thread t1 = new Thread(example::consume); t.start(); t1.start(); } }`

如上代码所示当队列的商品已经超过3个了则会阻塞生产者的线程不能再生产了，此时会阻塞线程，然后这个时候消费者就可以重新抢占到线程然后开始消费

#### Condition源码分析 ####

await源码

scss

复制代码

`public final void await() throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); //创建了一个新节点数据结构还是一个链表（如下图所示） Node node = addConditionWaiter(); //释放当前的锁并唤醒AQS队列中的一个线程 int savedState = fullyRelease(node); int interruptMode = 0; //如果当前结点不在AQS队列中则阻塞当前线程 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 // 如果node 的下一个等待者不是null, 则进行清理Condition 队列上的节点 unlinkCancelledWaiters(); if (interruptMode != 0) reportInterruptAfterWait(interruptMode); }`

中间最重要的代码是`while`循环,这里判断条件是当前的这个结点是不是在等待队列中，正常情况下肯定不是的，所以此时会阻塞当前线程等待被唤起

condition节点结构图

![format_png][]

addConditionWaiter

![format_png 1][]

这里很简单就是创建一个新的节点，然后维护节点链表，其中会将取消的节点清理掉

fullyRelease

ini

复制代码

`final int fullyRelease(Node node) { boolean failed = true; try { int savedState = getState(); if (release(savedState)) { failed = false; return savedState; } else { throw new IllegalMonitorStateException(); } } finally { if (failed) node.waitStatus = Node.CANCELLED; } }`

ini

复制代码

`protected final boolean tryRelease(int releases) { int c = getState() - releases; if (Thread.currentThread() != getExclusiveOwnerThread()) throw new IllegalMonitorStateException(); boolean free = false; if (c == 0) { free = true; setExclusiveOwnerThread(null); } setState(c); return free; }`

java

复制代码

`public final boolean release(int arg) { if (tryRelease(arg)) { Node h = head; if (h != null && h.waitStatus != 0) unparkSuccessor(h); return true; } return false; }`

就是将AQS的state置为0和`exclusiveOwnerThread`置为null,标识当前锁没有人占用然后就是`unpark-AQS`中的头节点就是队伍的第一个线程。

signal源码

java

复制代码

`public final void signal() { if (!isHeldExclusively()) throw new IllegalMonitorStateException(); Node first = firstWaiter; //找到第一个等待的节点 if (first != null) doSignal(first); }`

ini

复制代码

`private void doSignal(Node first) { do { if ( (firstWaiter = first.nextWaiter) == null) lastWaiter = null; first.nextWaiter = null; } while (!transferForSignal(first) && (first = firstWaiter) != null); }`

这里其实就是去除第一个节点然后再修改一下`condition`链表如下图所示

![format_png 2][]

transferForSignal

scss

复制代码

`final boolean transferForSignal(Node node) { /* * If cannot change waitStatus, the node has been cancelled. */ //这里就是将结点的等待状态修改成初始状态0 if (!compareAndSetWaitStatus(node, Node.CONDITION, 0)) return false; /* * Splice onto queue and try to set waitStatus of predecessor to * indicate that thread is (probably) waiting. If cancelled or * attempt to set waitStatus fails, wake up to resync (in which * case the waitStatus can be transiently and harmlessly wrong). */ //这里就是将等待的这个节点添加到AQS队列中 Node p = enq(node); int ws = p.waitStatus; //这里将node的上一个结点的等待状态设置成SIGNAL if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL)) //需要注意的是正常情况下是不在这里unpark的，而是放到AQS队列中等待unpark LockSupport.unpark(node.thread); return true; }`

这里很多人都会以为在这里执行了`LockSupport.unpark(node.thread);`所以唤醒了线程，其实不是，正常情况下还是要放到AQS队列中等待unpark，比如当前线程unlock了，如果刚好是当前线程可以抢占锁了此时就会`unpark`当前线程

ini

复制代码

`private Node enq(final Node node) { for (;;) { Node t = tail; if (t == null) { // Must initialize if (compareAndSetHead(new Node())) tail = head; } else { node.prev = t; if (compareAndSetTail(t, node)) { t.next = node; return t; } } } }`

#### Condition原理总结 ####

就是在维护两个队列配合使用`LockSupport.unpark(node.thread);`和`LockSupport.park(node.thread);`两个方法，具体逻辑图如下：

![format_png 3][]

作者：Potato\_土豆  
链接：https://juejin.cn/post/7233782463079006245

[format_png]: https://img-blog.csdnimg.cn/img_convert/6b36293d22c9ea91c0178ccaacfd8c21.webp?x-oss-process=image/format,png
[format_png 1]: https://img-blog.csdnimg.cn/img_convert/a4ec08798208cd858b9e07962f75cf56.webp?x-oss-process=image/format,png
[format_png 2]: https://img-blog.csdnimg.cn/img_convert/2bbe0b074de285fc9a7c5012a6925d7b.webp?x-oss-process=image/format,png
[format_png 3]: https://img-blog.csdnimg.cn/img_convert/bc6de56a8ed8597884c348994ff1520f.webp?x-oss-process=image/format,png