Java并发编程之Condition await/signal原理剖析

文章目录

Java并发编程之Condition await/signal原理剖析
- Condition与Lock的关系
- Condition实现原理
- await()实现分析
- signal()实现分析
- Condition接口与Object监听器的区别

Condition与Lock的关系

Condition本身也是⼀个接口，其功能和wait/notify类似，如下所示：

public interface Condition {
	void await() throws InterruptedException;
	boolean await(long time, TimeUnit unit) throws InterruptedException;
	long awaitNanos(long nanosTimeout) throws InterruptedException;
	void awaitUninterruptibly();
	boolean awaitUntil(Date deadline) throws InterruptedException;
	void signal();
	void signalAll();
}

wait()/notify()必须和synchronized⼀起使用， Condition也必须和Lock⼀起使用。因此，在Lock的接口中，有⼀个与Condition相关的接口：

public interface Lock {
	void lock();
	void lockInterruptibly() throws InterruptedException;
	// 所有的Condition都是从Lock中构造出来的
	Condition newCondition();
	boolean tryLock();
	boolean tryLock(long time, TimeUnit unit) throws InterruptedException;
	void unlock();
}

Condition实现原理

可以发现， Condition的使用很方便，避免了wait/notify的生产者通知生产者、消费者通知消费者的问题。由于Condition必须和Lock⼀起使用，所以Condition的实现也是Lock的⼀部分。首先查看互斥锁和读写锁中Condition的构造方法：

public class ReentrantLock implements Lock, java.io.Serializable {
    // ...
    public Condition newCondition() {
        return sync.newCondition();
    }
}
public class ReentrantReadWriteLock
        implements ReadWriteLock, java.io.Serializable {
    // ...
    private final ReentrantReadWriteLock.ReadLock readerLock;
    private final ReentrantReadWriteLock.WriteLock writerLock;
    // ...
    public static class ReadLock implements Lock, java.io.Serializable {
        // 读锁不⽀持Condition
        public Condition newCondition() {
// 抛异常
            throw new UnsupportedOperationException();
        }
    }
    public static class WriteLock implements Lock, java.io.Serializable {
        // ...
        public Condition newCondition() {
            return sync.newCondition();
            }
			// ...
        }
		// ...
}

首先，读写锁中的 ReadLock 是不⽀持 Condition 的，读写锁的写锁和互斥锁都⽀持Condition。虽然它们各自调用的是自己的内部类Sync，但内部类Sync都继承自AQS。因此，上面的代码sync.newCondition最终都调用了AQS中的newCondition：

public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer
        implements java.io.Serializable {
    public class ConditionObject implements Condition, java.io.Serializable {
		// Condition的所有实现，都在ConditionObject类中
    }
}
abstract static class Sync extends AbstractQueuedSynchronizer {
    final ConditionObject newCondition() {
        return new ConditionObject();
    }
}

有上面代码可知，调用 Lock.newConditon() 方法实际是调用的 Sync内部类中的方法创建了Condition实现了 ConditionObject()。ConditionObject 类是同步器 AQS 的内部类，因为 Condition 的操作需要相关联的锁，每一个Condition对象上面，都阻塞了多个线程。因此，在 ConditionObject 内部也有一个双向链表组成的队列，如下所示：

在这里插入图片描述

public class ConditionObject implements Condition, java.io.Serializable {
    private transient Node firstWaiter;
    private transient Node lastWaiter;
}
static final class Node {
    volatile Node prev;
    volatile Node next;
    volatile Thread thread;
    Node nextWaiter;
}

下面来看⼀下在await()/notify()方法中，是如何使用这个队列的。

await()实现分析

    public final void await() throws InterruptedException {
		// 刚要执⾏await()操作，收到中断信号，抛异常
        if (Thread.interrupted())
            throw new InterruptedException();
		// 加⼊Condition的等待队列
        Node node = addConditionWaiter();
		// 阻塞在Condition之前必须先释放锁，否则会死锁
        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;
        if (node.nextWaiter != null) // clean up if cancelled
            unlinkCancelledWaiters();
        if (interruptMode != 0)
			// 被中断唤醒，抛中断异常
            reportInterruptAfterWait(interruptMode);
    }

由上面的源码可知，如果一个线程调用了Condition.await()方法，那么该线程将会释放锁、构造成节点加入等待队列并进入等待状态。当调用await()方法时，相当于同步队列的首节点（获取了锁的节点）移动到Condition的等待队列中。

在这里插入图片描述

关于await，有几个关键点要说明：

线程调用await()的时候，肯定已经先拿到了锁。所以，在 addConditionWaiter()内部，对这个双向链表的操作不需要执行CAS操作，线程天生是安全的，代码如下：

    private Node addConditionWaiter() {
		// ...
        Node t = lastWaiter;
        Node node = new Node(Thread.currentThread(), Node.CONDITION);
        if (t == null)
            firstWaiter = node;
        else
            t.nextWaiter = node;
        lastWaiter = node;
        return node;
    }

在线程执行wait操作之前，必须先释放锁。也就是fullyRelease(node)，否则会发⽣死锁。这个和wait/notify与synchronized的配合机制⼀样。
线程从wait中被唤醒后，必须用acquireQueued(node, savedState)方法重新拿锁。线程从wait中被唤醒后，只是从等待队列转移到同步队列中，仍然需要在同步队列中排队争取锁。
checkInterruptWhileWaiting(node)代码在park(this)代码之后，是为了检测在park期间是否收到过中断信号。当线程从park中醒来时，有两种可能：一种是其他线程调用了unpark，另⼀种是收到中断信号。这里的await()方法是可以响应中断的，所以当发现自己是被中断唤醒的，而不是被unpark唤醒的时，会直接退出while循环， await()方法也会返回。
isOnSyncQueue(node)用于判断该Node是否在AQS的同步队列里面。初始的时候， Node只在Condition的队列里，而不在AQS的队列里，但执行notity操作的时候，会放进AQS的同步队列。

signal()实现分析

    public final void signal() {
		// 只有持有锁的线程，才有资格调⽤signal()⽅法
        if (!isHeldExclusively())
            throw new IllegalMonitorStateException();
        Node first = firstWaiter;
        if (first != null)
			// 发起通知
            doSignal(first);
    }

    // 唤醒队列中的第1个线程
    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) {
        if (!node.compareAndSetWaitStatus(Node.CONDITION, 0))
        	return false;
		// 先把Node放⼊互斥锁的同步队列中，再调⽤unpark⽅法
        Node p = enq(node);
        int ws = p.waitStatus;
        if (ws > 0 || !p.compareAndSetWaitStatus(ws, Node.SIGNAL))
        LockSupport.unpark(node.thread);
        	return true;
    }