Condition类及其子类的源码讲解_if (ws > 0 || !compareandsetwaitstatus(p, ws, node

作者：很楠不爱3 | 2024-06-09 06:31:35

踩

if (ws > 0 || !compareandsetwaitstatus(p, ws, node.signal)) locksupport.unpa

Condition类及其子类的源码讲解

源码作者：Doug Lea
文章作者：我

上一期讲完了AQS类的源码，其中还包含一个ConditionObject类，是Condition的子类，AQS和Condition都是存放线程的队列，所以大家总是把两者对比来讲，所以对于Condition，我单开一篇博客给大家解析，以下内容除源码，其他纯手打(包括所有注释)，希望对大家有帮助哈！！！

在解析之前，大家得先了解AQS和Condition的区别：

AQS是一个同步队列，AQS队列中的节点等待资源释放来被唤醒；Condition是一个条件队列，Condition中的节点等待singal()/signalAll()信号的唤醒。
AQS队列是一个双向链表结构，Condition是一个单向链表结构。
Condition对节点进行await阻塞操作时，必须保证该节点已经在AQS的同步队列中阻塞，否则在fullyRelease()中会抛出异常；Condition进行signal()或者signalAll()唤醒操作时，唤醒的只能是在Condition队列阻塞的节点，唤醒后的节点会在AQS队列阻塞或者直接拿到锁执行。

下面的解析中，我就把AQS队列记作同步队列，Condition队列记作条件队列。

Condition类

public interface Condition {
	// 当前线程从AQS中阻塞进入Condition中等待
	// 使用该方法等待的线程支持通过中断的方式被唤醒
    void await() throws InterruptedException;
    // 当前线程从AQS中阻塞进入Condition中等待
    // 使用该方法等待的线程不支持通过中断的方式被唤醒，只能通过signal和signalAll唤醒
    void awaitUninterruptibly();
    // 支持设置超时的等待，等待指定纳秒后自动被唤醒，也可支持中断唤醒
    // 返回的值如果大于0说明等待时间未超时；如果小于等于0，说明等待超时，自动被唤醒
    long awaitNanos(long nanosTimeout) throws InterruptedException;
    // 支持设置超时的等待，等待指定单位的指定时间后自动被唤醒，也可支持中断唤醒
    // 返回flase就是等待超时的意思，自动被唤醒；true就是等待时间未超时，还可继续等待
    boolean await(long time, TimeUnit unit) throws InterruptedException;
    // 支持设置超时的等待，等待到达设置的截止日期后自动被唤醒，也可支持中断唤醒
    // 返回flase就是等待超时的意思，自动被唤醒；true就是等待时间未超时，还可继续等待
    boolean awaitUntil(Date deadline) throws InterruptedException;
    // 唤醒条件队列中第一个正在等待的节点
    void signal();
    // 唤醒条件队列中所有正在等待的节点
    void signalAll();
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

ConditionObject类

解析这个类时我不想把这个类所有的属性和方法挨个打注释，我感觉没意思，看客老爷们肯定也是。所以我打算放到实际的应用中，由外到内整体解析JUC锁类使用newCondition时的await()阻塞方法和signal()/signalAll()唤醒方法。

ConditionObject类属性解析

public class ConditionObject implements Condition, java.io.Serializable {
	// 存放条件队列头节点，方便出队操作
    private transient Node firstWaiter;
    // 存放条件队列尾节点，方便入队操作
    private transient Node lastWaiter;
    // 表明该节点在条件队列唤醒之后被中断
    private static final int REINTERRUPT =  1;
    // 表明该节点在条件队列等待过程中被中断
    private static final int THROW_IE    = -1;
}
1
2
3
4
5
6
7
8
9
10

await()：条件队列阻塞操作

public final void await() throws InterruptedException {
	// 如果当前线程被中断，抛出中断异常
    if (Thread.interrupted())
        throw new InterruptedException();
    // 创建新的Condition节点入条件队列
    Node node = addConditionWaiter();
    // 释放同步队列锁成功后，要保存已释放锁的同步状态，因为该节点唤醒之后要把同步状态恢复
    int 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;
    // 如果节点在条件队列的后继节点不为null，那就清除条件队列中无效的节点
    if (node.nextWaiter != null) 
        unlinkCancelledWaiters();
    // 如果该节点已被中断，根据当前中断模式判断是应该中断线程还是抛出异常
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30

await()整体流程：

一个线程拿到锁后，调用此方法，先在条件队列尾插一个新的装有当前线程任务的节点，然后释放同步队列中的头节点(当前锁节点)。
通过while循环判断尾插的新条件队列的节点是否在同步队列上，如果不在，就一直阻塞这个线程；如果在同步队列，就说明发生了【被signal/signalAll唤醒】或者【被中断】，并且开始尝试获取同步队列中的锁，获取到就直接执行，获取不到就尾插到同步队列。

在这个方法中发生了节点转移(同步队列–>条件队列)：
当时我死活想不明白在哪里转移的，我认为就没有发生节点的转移，只是通过一个while循环来实现阻塞这个线程，直到【被signal/signalAll唤醒】或者【被中断】才停止阻塞，但如果这样，那创建一个新的条件队列的节点就没有意义了，最后通过黄师傅的点拨，也是因为自己的粗心，没有注意到addConditionWaiter方法中的
Node node = new Node(Thread.currentThread(), Node.CONDITION);
就是把当前线程任务放到了新创建的这个节点，最终才恍然大悟。

在这里还是提醒大家看源码首先要细心，然后如果想不明白出去换换脑子可能就会来灵感了。

addConditionWaiter()

private Node addConditionWaiter() {
	// 获取Condition尾节点快照
    Node t = lastWaiter;
    // 如果尾节点不为null并状态不是条件队列中的正常状态
    if (t != null && t.waitStatus != Node.CONDITION) {
    	// 整理条件队列，清除无效节点
        unlinkCancelledWaiters();
        // 重新获取尾节点
        t = lastWaiter;
    }
    // 为当前线程创建一个状态为CONDITION的新节点
    Node node = new Node(Thread.currentThread(), Node.CONDITION);
    // 如果尾节点为null，说明，条件队列为null，那就把新节点设置为条件队列的头结点
    if (t == null)
        firstWaiter = node;
    else
    	// 如果条件队列不为空，那就让尾节点指向新节点
        t.nextWaiter = node;
    // 不管条件队列是不是为null，新节点都是尾节点，把新节点设置为尾节点
    lastWaiter = node;
    // 返回新节点
    return node;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

unlinkCancelledWaiters()

// 循环消除条件队列中的无效节点（下方有详解）
private void unlinkCancelledWaiters() {
	// 存放队列遍历的初始阶段-队列的头结点
    Node t = firstWaiter;
    // 记录循环中的当前节点
    Node trail = null;
    while (t != null) {
    	// 获取到当前节点的后继结点
        Node next = t.nextWaiter;
        // 如果循环中当前节点不为null，并且当前节点是无效状态
        if (t.waitStatus != Node.CONDITION) {
        	// 把无效的当前节点指向null，便于垃圾回收
            t.nextWaiter = null;
            /* 
	            如果当前节点是null，只有一种情况，那就是第一次循环才会出现trail == null，
	            如果这个条件成立，那就把后继结点变为头结点
            */
            if (trail == null)
                firstWaiter = next;
            else
            	// 如果不是第一次循环，那就让当前节点的后继指针指向当前节点的后继结点
                trail.nextWaiter = next;
            /* 
	            如果当前节点的后继结点为null，说明当前节点就是条件队列中最后一个节点了，
	            那就把当前节点设置为尾节点
            */
            if (next == null)
                lastWaiter = trail;
        }
        else
        	// 如果当前节点是正常状态，那就把当前结点赋值给trail
            trail = t;
        // 将条件队列的后继节点赋值给t
        t = next;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36

我感觉这个方法其实是最难理解的，但是看懂了之后却是最容易的，它要做的就是为了整理条件队列中的节点，从头到尾循环条件队列的所有节点，把无效节点剔除，始终保持条件队列中各个节点的有效性。

fullyRelease()

final int fullyRelease(Node node) {
    // 创建入队失败标识
	boolean failed = true;
	try {
		/* 
			获取同步的状态-信号量，为0，说明没有加锁，即没有在同步队列阻塞，
			大于0，表示的是锁重入的数量
		*/
	    int savedState = getState();
	    // 执行同步队列中释放并唤醒锁操作
	    if (release(savedState)) {
	    	// 入队成功并返回同步状态
	        failed = false;
	        return savedState;
	    } else {
	    	// 如果释放锁失败，抛异常
	        throw new IllegalMonitorStateException();
	    }
	} finally {
		// 如果在同步队列中释放锁失败，那就把新节点设置为无效状态
	    if (failed)
	        node.waitStatus = Node.CANCELLED;
	}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

release()

public final boolean release(int arg) {
	// 调用子类方法实现释放同步队列中的锁
    if (tryRelease(arg)) {
        Node h = head;
        // 如果释放成功，同步队列中头节点存在并还是初始状态，那就唤醒头节点的后继节点
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}
1
2
3
4
5
6
7
8
9
10
11

isOnSyncQueue()

final boolean isOnSyncQueue(Node node) {
	/* 
		如果创建的新节点状态为CONDITION ，说明该节点在条件队列；
		因为条件队列是单向链表，所以如果前驱结点不为null，说明该节点在条件队列。
	*/
    if (node.waitStatus == Node.CONDITION || node.prev == null)
        return false;
    // 如果节点后继节点不为null，那就说明在同步队列，(Node类的next属性是用于同步队列的)
    if (node.next != null) 
        return true;
    // 在同步队列从后往前遍历，判断这个节点到底在哪个队列上
    return findNodeFromTail(node);
}
1
2
3
4
5
6
7
8
9
10
11
12
13

findNodeFromTail()

private boolean findNodeFromTail(Node node) {
	// 获取到同步队列的尾节点快照
    Node t = tail;
    for (;;) {
    	// 如果新创建的Condition节点是同步队列的尾节点，那就返回true
        if (t == node)
            return true;
        // 如果尾节点是null，说明同步队列为空队列，那就返回false
        if (t == null)
            return false;
        // 如果以上条件都不成立，那就检查同步队列尾节点的前驱结点，并重复以上操作
        t = t.prev;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14

checkInterruptWhileWaiting()

private int checkInterruptWhileWaiting(Node node) {
	// 先判断当前线程是否中断状态，如果不是就返回0
    return Thread.interrupted() ?
    	// 如果当前线程加入同步队列成功，那就返回THROW_IE ，否则，说明线程阻塞，返回REINTERRUPT
        (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :
        0;
}
1
2
3
4
5
6
7

transferAfterCancelledWait()

final boolean transferAfterCancelledWait(Node node) {
	// 先将该节点状态由CONDITION修改为0初始化状态
    if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {
    	// 如果CAS成功，就执行同步队列入队操作
        enq(node);
        return true;
    }
    // 如果判断节点不在同步队列上，那就阻塞当前线程，由执行状态变为就绪状态
    while (!isOnSyncQueue(node))
        Thread.yield();
    return false;
}
1
2
3
4
5
6
7
8
9
10
11
12

enq()

private Node enq(final Node node) {
    for (;;) {
    	// 先存储同步队列队尾节点的快照
        Node t = tail;
        if (t == null) {
        	// 如果没有队尾节点，说明队列中没有结点，那就创建一个空节点并CAS设置为队首结点
            if (compareAndSetHead(new Node()))
            	/* 
	            	CAS成功后，因为队列现在只有一个节点，
	            	所以队首和队尾节点都设置为当前这个新的节点，并执行死循环逻辑
            	*/
                tail = head;
        } else {
        	/**
	        	如果队尾指针有值，说明当前队列不是空队列，
	        	那就把新创建的共享锁的node结点的前驱指针指向队列的队尾结点
        	*/
            node.prev = t;
            /* 
		        为防止有其他现成已经修改了队尾节点了，
		        所以使用CAS来把队列队尾的结点由快照获取的t修改为新创建的node
	        */
            if (compareAndSetTail(t, node)) {
            	// 如果CAS替换队尾节点成功，那就把原队尾结点t的后继指针指向新创建的node节点
        		// 并返回原队尾结点t
                t.next = node;
                return t;
            }
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

reportInterruptAfterWait()

private void reportInterruptAfterWait(int interruptMode)
    throws InterruptedException {
    // 如果是在条件队列等待过程中被中断，那就抛出中断异常
    if (interruptMode == THROW_IE)
        throw new InterruptedException();
    // 如果实在条件队列中唤醒后被中断，那就让当前线程中断，阻塞节点
    else if (interruptMode == REINTERRUPT)
        selfInterrupt();
}
1
2
3
4
5
6
7
8
9

signal()：条件队列唤醒单任务操作

public final void signal() {
	// 子类调用判断当前线程是否上锁，如果没有上锁，那就抛出异常
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    // 获得条件队列头节点快照
    Node first = firstWaiter;
    // 头节点不为null，执行唤醒头节点逻辑
    if (first != null)
        doSignal(first);
}
1
2
3
4
5
6
7
8
9
10

doSignal()

private void doSignal(Node first) {
    do {
    	// 如果条件队列头节点的后继结点为null，那就设置条件队列尾节点为null
        if ( (firstWaiter = first.nextWaiter) == null)
            lastWaiter = null;
        // 将头节点的后继指针设为null，使头节点在条件队列中移除
        first.nextWaiter = null;
    // 如果唤醒操作执行失败并且头节点不是null，那就循环操作，直到唤醒单个节点成功
    } while (!transferForSignal(first) &&
             (first = firstWaiter) != null);
}
1
2
3
4
5
6
7
8
9
10
11

在这个方法中又发生了节点转移(条件队列–>同步队列)：
与上一次转移不同的是，这次先把条件队列中要唤醒的节点释放，然后把这个释放后的节点快照尾插到同步队列阻塞或者直接拿到锁执行。

transferForSignal()

final boolean transferForSignal(Node node) {
	// CAS节点状态由CONDITION修改为初始状态0
    if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
    	// 返回唤醒失败标识
        return false;
    // CAS成功，执行node节点同步队列入队操作，并返回node节点的前驱节点
    Node p = enq(node);
    int ws = p.waitStatus;
    /* 
	    如果node的前驱节点为无效节点，或者修改前驱节点状态为SIGNAL失败，
	    那就唤醒入队后的node节点执行任务
    */
    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
        LockSupport.unpark(node.thread);
    // 返回唤醒成功标识
    return true;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17

signalAll()：条件队列唤醒全部任务操作

public final void signalAll() {
    // 子类调用判断当前线程是否上锁，如果没有上锁，那就抛出异常
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    // 获得条件队列头节点快照
    Node first = firstWaiter;
    // 头节点不为null，执行唤醒所有节点逻辑
    if (first != null)
        doSignalAll(first);
}
1
2
3
4
5
6
7
8
9
10

doSignalAll()

private void doSignalAll(Node first) {
	// 在条件队列从头结点到尾节点循环唤醒节点，直到唤醒所有等待节点
    lastWaiter = firstWaiter = null;
    do {
        Node next = first.nextWaiter;
        first.nextWaiter = null;
        transferForSignal(first);
        first = next;
    } while (first != null);
}
1
2
3
4
5
6
7
8
9
10

对于signal/signalAll唤醒方法，两个唯一差别就是：
single方法只唤醒一个当前节点的有效后继节点。
signalAll方法会在条件队列从头往尾循环唤醒所有有效的节点。

本篇博客要是有什么技术问题，欢迎大家来指正。

如果大家有什么不理解的，也欢迎大家评论，我很高兴和大家一起讨论技术问题。

声明：本文内容由网友自发贡献，转载请注明出处：【wpsshop博客】