Condition接口、LockSupport工具类

清疚 2022-02-25 07:14 261阅读 0赞

## **Condition介绍** ##

任意一个Java对象，都拥有一组监视器方法（定义在java.lang.Object中），主要包括**`wait()`、`wait(long timeout)`、`notify()`**以及**`notifyAll()`**方法，这些方法与synchronized同步关键字配合，可以实现等待/通知模式。Condition接口也提供了类似Object的监视器方法，与Lock配合可以实现等待/通知模式，但是这两者在使用方式以及功能特性上还是有差别的。Object上的监视器只有一个等待队列，但是Condition上可以有多个等待队列，每个`Lock.newCondition`都可以产生一个Condition对象，每个Condition对象关联到一个新的等待队列上。并且Condition支持响应中断，但是Object上的监视器不支持。

一个lock可以产生多个Condition对象。

## **Condition接口示例** ##

Condition定义了等待/通知两种类型的方法，当前线程调用这些方法时，需要提前获取到Condition对象关联的锁。Condition对象是由Lock对象（调用Lock对象的newCondition()方法）创建出来的，换句话说，Condition是依赖Lock对象的。

一般都会将Condition对象作为成员变量。当调用await()方法后，当前线程会释放锁并在此等待，而其他线程调用Condition对象的signal()方法，通知当前线程后，当前线程才从await()方法返回，并且在返回前已经获取了锁。

public interface Condition {
    
         // 当前线程进入等待状态直到被通知(signal)或被中断
        void await() throws InterruptedException;
        // 不响应中断等待，直到被通知(signal)
        void awaitUninterruptibly();
        // 等待指定时长直到被通知或中断或超时。
        long awaitNanos(long nanosTimeout) throws InterruptedException;
        // 等待指定时长直到被通知或中断或超时。
        boolean await(long time, TimeUnit unit) throws InterruptedException;
        // 当前线程进入等待状态直到被通知、中断或者到某个时间。如果没有到指定事件就被通知，方法返回true，否则false。 
        boolean awaitUntil(Date deadline) throws InterruptedException;
        // 唤醒一个等待在Condition上的线程，该线程从等待方法返回前必须获得与Condition相关联的锁  
        void signal();
        // 唤醒所有等待在Condition上的线程，该线程从等待方法返回前必须获得与Condition相关联的锁  
        void signalAll();
    }

## **Condition实现原理分析** ##

ConditionObject是同步器AbstractQueuedSynchronizer的内部类，因为Condition的操作需要获取相关联的锁，所以作为同步器的内部类也较为合理。每个Condition对象都包含着一个队列（以下称为等待队列），该队列是Condition对象实现等待/通知功能的关键。

下面将分析Condition的实现，主要包括：等待队列、等待和通知，下面提到的Condition如果不加说明均指的ConditionObject。

### 等待队列 ###

等待队列是一个FIFO的队列，在队列中的每个节点都包含了一个线程引用，该线程就是在Condition对象上等待的线程，如果一个线程调用了Condition.await()方法，那么该线程将会释放锁、构造成节点加入等待队列并进入等待状态。事实上，节点的定义复用了同步器中节点的定义，也就是说，同步队列和等待队列中节点类型都是同步器的静态内部类`AbstractQueuedSynchronizer.Node`。

一个Condition包含一个等待队列，Condition拥有首节点（firstWaiter）和尾节点（lastWaiter）。当前线程调用`Condition.await()`方法，将会以当前线程构造节点，并将节点从尾部加入等待队列，等待队列的基本结构下

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0ppblhZYW4_size_16_color_FFFFFF_t_70][]

如图所示，Condition拥有首尾节点的引用，而新增节点只需要将原有的尾节点nextWaiter指向它，并且更新尾节点即可。上述节点引用更新的过程并没有使用CAS保证，原因在于调用await()方法的线程必定是获取了锁的线程，也就是说该过程是由锁来保证线程安全的。所以读锁时没有Condition的。

在Object的监视器模型上，一个对象拥有一个**同步队列和等待队列**，**而并发包中的Lock（更确切地说是同步器）拥有一个同步队列和多个等待队列**，其对应关系如下图所示。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0ppblhZYW4_size_16_color_FFFFFF_t_70 1][]

如图所示，Condition的实现是同步器的内部类，因此每个Condition实例都能够访问同步器提供的方法，相当于每个Condition都拥有所属同步器的引用。

public final void await() throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
            
        // 将当前线程构造成一个node，加入到wait队列中。
        Node node = addConditionWaiter(); 
        //最终调用AQS的release释放同步锁
        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);
    }

从上面的代码流程，我们知道主流程应该是：当一个线程调用`await()`后，首先将当前线程构造成一个node节点，加入到当前ConditionObject实例对象的等待队列尾部，释放当前线程占用的锁，然后阻塞等待另外线程`unpark`唤醒当前线程，当前线程唤醒后尝试获取同步状态，获取成功返回，继续往下走。

### 通知 ###

调用Condition的signal()方法，将会唤醒在等待队列中等待时间最长的节点（首节点），在唤醒节点之前，会将节点移到同步队列中。

Condition的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的queue中节点移到同步队列中
    // Transfers a node from a condition queue onto sync queue.
    final boolean transferForSignal(Node node) {
    
        if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
            return false;
        // 将当前node加入到同步队列中，等待获取锁
        Node p = enq(node);
        int ws = p.waitStatus;
        //告诉前驱节点，在释放状态是通知自己。
        if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
            LockSupport.unpark(node.thread);
        return true;
    }

调用该方法的前置条件是当前线程必须获取了锁，可以看到`signal()`方法进行了`isHeldExclusively()`检查，也就是当前线程必须是获取了锁的线程。接着获取等待队列的首节点，将其移动到同步队列并使用LockSupport唤醒节点中的线程。

当前线程加入等待队列的流程图：

![image.png-165.6kB][]

节点从等待队列移动到同步队列：

![image.png-131.5kB][]

通过调用同步器的`enq(Node node)`方法，等待队列中的头节点线程安全地移动到同步队列。当节点移动到同步队列后，当前线程再使用LockSupport唤醒该节点的线程。

被唤醒后的线程，将从await()方法中的while循环中退出（`isOnSyncQueue(Node node`)方法返回true，节点已经在同步队列中），进而调用同步器的`acquireQueued()`方法加入到获取同步状态的竞争中。

成功获取同步状态（或者说锁）之后，被唤醒的线程将从先前调用的await()方法返回，此时该线程已经成功地获取了锁。

Condition的`signalAll()`方法，相当于对等待队列中的每个节点均执行一次signal()方法，效果就是将等待队列中所有节点全部移动到同步队列中，并唤醒每个节点的线程。

## **实例** ##

Conditon使用例子如下，可以实现条件性的通知

public class WaitNotify {
    
        static ReentrantLock lock = new ReentrantLock();
        static Condition conditionA  = lock.newCondition();
        static Condition conditionB = lock.newCondition();
    
        public static void main(String[] args) throws InterruptedException {
            Thread waitThreadA = new Thread(new WaitA(), "WaitThreadA");
            waitThreadA.start();
            Thread waitThreadB = new Thread(new WaitB(), "WaitThreadB");
            waitThreadB.start();
            TimeUnit.SECONDS.sleep(2);
            lock.lock();
            try {
                conditionA.signal();
            } finally {
                lock.unlock();
            }
        }
    
        static class WaitA implements Runnable {
    
            @Override
            public void run() {
                lock.lock();
                try {
                    System.out.println(Thread.currentThread() + " begin await @ "
                            + new SimpleDateFormat("HH:mm:ss").format(new Date()));
                    conditionA.await();
                    System.out.println(Thread.currentThread() + " end await @ "
                            + new SimpleDateFormat("HH:mm:ss").format(new Date()));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    lock.unlock();
                }
            }
        }
    
        static class WaitB implements Runnable {
    
            @Override
            public void run() {
                lock.lock();
                try {
                    System.out.println(Thread.currentThread() + " begin await @ "
                            + new SimpleDateFormat("HH:mm:ss").format(new Date()));
                    conditionB.await();
                    System.out.println(Thread.currentThread() + " end await @ "
                            + new SimpleDateFormat("HH:mm:ss").format(new Date()));
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    lock.unlock();
                }
            }
        }
    }
    结果：
    Thread[WaitThreadA,5,main] begin await @ 00:49:57 
    Thread[WaitThreadB,5,main] begin await @ 00:49:57 
    Thread[WaitThreadA,5,main] end await @ 00:49:59

WaitThreadB因为没有被通知，一直阻塞。**这里说一下Condition的大概实现，AbstractQueuedSynchronizer内部维护着一个同步队列（双向链表实现），多个条件队列（单向链表实现），条件队列由AbstractQueuedSynchronizer的内部类ConditionObject来维护，new一个ConditonObject ，则多一个条件队列，当一个线程执行await方法是，会把当线程包装成一个Node节点，放到执行await方法的ConditionObject的条件队列中，释放锁并被阻塞，当执行signal方式时，会把条件队列的第一个节点移除，并转移到同步队列中，获取到锁即可继续执行。signalall则是将队列中所有的节点都唤醒加入到同步队列中去。**

## **LockSupport工具类** ##

**1、简单**

俗话说，没有比较就没有伤害。这里咱们还是通过对比来介绍LockSupport的简单。

在没有LockSupport之前，线程的挂起和唤醒咱们都是通过Object的wait和notify/notifyAll方法实现。

写一段例子代码，线程A执行一段业务逻辑后调用wait阻塞住自己。主线程调用notify方法唤醒线程A，线程A然后打印自己执行的结果。

public class TestObjWait {
    
        public static void main(String[] args)throws Exception {
            final Object obj = new Object();
            Thread A = new Thread(new Runnable() {
                @Override
                public void run() {
                    int sum = 0;
                    for(int i=0;i<10;i++){
                        sum+=i;
                    }
                    try {
                        synchronized (obj){
                            obj.wait();
                        }
                    }catch (Exception e){
                        e.printStackTrace();
                    }
                    System.out.println(sum);
                }
            });
            A.start();
            //睡眠一秒钟，保证线程A已经计算完成，阻塞在wait方法
            Thread.sleep(1000);
            synchronized (obj){
                obj.notify();
            }
        }
    }

那如果咱们换成LockSupport呢？简单得很，看代码：

public class TestObjWait {
    
        public static void main(String[] args)throws Exception {
            Thread A = new Thread(new Runnable() {
                @Override
                public void run() {
                    int sum = 0;
                    for(int i=0;i<10;i++){
                        sum+=i;
                    }
                    LockSupport.park();
                    System.out.println(sum);
                }
            });
            A.start();
            //睡眠一秒钟，保证线程A已经计算完成，阻塞在wait方法
            Thread.sleep(1000);
            LockSupport.unpark(A);
        }
    }

**2、灵活**

如果只是LockSupport在使用起来比Object的wait/notify简单，那还真没必要专门讲解下LockSupport。最主要的是灵活性。

上边的例子代码中，主线程调用了Thread.sleep(1000)方法来等待线程A计算完成进入wait状态。如果去掉Thread.sleep()调用，代码如下：

public class TestObjWait {
    
        public static void main(String[] args)throws Exception {
            final Object obj = new Object();
            Thread A = new Thread(new Runnable() {
                @Override
                public void run() {
                    int sum = 0;
                    for(int i=0;i<10;i++){
                        sum+=i;
                    }
                    try {
                        synchronized (obj){
                            obj.wait();
                        }
                    }catch (Exception e){
                        e.printStackTrace();
                    }
                    System.out.println(sum);
                }
            });
            A.start();
            //睡眠一秒钟，保证线程A已经计算完成，阻塞在wait方法
            //Thread.sleep(1000);
            synchronized (obj){
                obj.notify();
            }
        }
    }

多运行几次上边的代码，有的时候能够正常打印结果并退出程序，但有的时候线程无法打印结果阻塞住了。原因就在于：主线程调用完notify后，线程A才进入wait方法，导致线程A一直阻塞住。由于线程A不是后台线程，所以整个程序无法退出。

那如果换做LockSupport呢？LockSupport就支持主线程先调用unpark后，线程A再调用park而不被阻塞吗？是的，没错。代码如下：

public class TestObjWait {
    
        public static void main(String[] args)throws Exception {
            final Object obj = new Object();
            Thread A = new Thread(new Runnable() {
                @Override
                public void run() {
                    int sum = 0;
                    for(int i=0;i<10;i++){
                        sum+=i;
                    }
                    LockSupport.park();
                    System.out.println(sum);
                }
            });
            A.start();
            //睡眠一秒钟，保证线程A已经计算完成，阻塞在wait方法
            //Thread.sleep(1000);
            LockSupport.unpark(A);
        }
    }

**总结一下，LockSupport比Object的wait/notify有两大优势**：

**①LockSupport不需要在同步代码块里 。所以线程间也不需要维护一个共享的同步对象了，实现了线程间的解耦。**

**②unpark函数可以先于park调用，所以不需要担心线程间的执行的先后顺序。**

参考文章：

[https://www.cnblogs.com/qingquanzi/p/8228422.html][https_www.cnblogs.com_qingquanzi_p_8228422.html]

[https://blog.csdn.net/zzti\_erlie/article/details/80358564][https_blog.csdn.net_zzti_erlie_article_details_80358564]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0ppblhZYW4_size_16_color_FFFFFF_t_70]: /images/20220225/ae0a7c2077734311b4aa59f12fe5a288.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0ppblhZYW4_size_16_color_FFFFFF_t_70 1]: /images/20220225/caae626afc1c4463914a9bfd79aff0da.png
[image.png-165.6kB]: /images/20220225/1e331c8a398448898680624d80a3e018.png
[image.png-131.5kB]: /images/20220225/1680b367d6c44027adbac2a75e06bd22.png
[https_www.cnblogs.com_qingquanzi_p_8228422.html]: https://www.cnblogs.com/qingquanzi/p/8228422.html
[https_blog.csdn.net_zzti_erlie_article_details_80358564]: https://blog.csdn.net/zzti_erlie/article/details/80358564