并发编程-Condition底层设计

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

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

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

Condition使用示例

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源码

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节点结构图

addConditionWaiter

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

fullyRelease

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;
    }
}

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;
}

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源码

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);
}

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

transferForSignal

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当前线程

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);两个方法，具体逻辑图如下：