[Java源码][并发J.U.C]---解析Condition

作者: nicktming | 来源:发表于2018-09-07 00:38 被阅读3次

前言

任意一个Java对象，都拥有一组监视器方法（定义在java.lang.Object上），主要包括wait(),wait(long timeout),notify()以及notifyAll()方法，这些方法与synchronized同步关键字配合，可以实现等待/通知模式。Condition接口也提供了类似Object的监视器方法,与Lock配合可以实现等待/通知模式.

本文代码: 代码下载

Object的监视器方法与Condition接口的对比

comparison.png

例子1

启动了一个线程并且生成了一个Condition对象con, 在线程中启动con.await()方法随后在主线程中通过con.signal()方法唤醒该线程.

package com.sourcecode.reentrantreadwritelock;

import java.util.concurrent.TimeUnit;
public class TestCondition {
    public static void main(String[] args) throws InterruptedException {
        new MyThread().start();
        TimeUnit.SECONDS.sleep(5);
        lock.lock();
        System.out.println(getPrefix() + " gets lock");
        System.out.println(getPrefix() + "try to signal");
        con.signal();
        System.out.println(getPrefix() + "after signal");
        System.out.println(getPrefix() + " releases lock");
        lock.unlock();
    }

    static String getPrefix() {
        return Thread.currentThread().getName() + "==============";
    }

    static ReentrantLock lock = new ReentrantLock();
    static Condition con = lock.newCondition();

    static class MyThread extends Thread {
        public void run() {
            lock.lock();
            System.out.println(getPrefix() + "gets first lock");
            lock.lock();
            System.out.println(getPrefix() + "gets second lock");
            try {
                System.out.println(getPrefix() + "invoke await...");
                con.await();
                System.out.println(getPrefix() + "after await...");
                //TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                lock.unlock();
                System.out.println(getPrefix() + "release first lock");
                lock.unlock();
                System.out.println(getPrefix() + "release second lock");
                //lock.unlock();
                //System.out.println("release third lock");
            }
        }
    }
}

结果如下: 跟预期没有什么区别.

Thread-0==============gets first lock
Thread-0==============gets second lock
Thread-0==============invoke await...
main============== gets lock
main==============try to signal
main==============after signal
main============== releases lock
Thread-0==============after await...
Thread-0==============release first lock
Thread-0==============release second lock

因此接下来看看如何实现的.

实现思路

如下图所示,Condition接口定义所有的API方法,具体的实现类是AbstractQueuedSynchronizer的内部类ConditionObject,为什么要把实现类放到AbstractQueuedSynchronizer类中, 是因为ConditionObject需要用到该类中的结构比如Node类. 如果对于AbstractQueuedSynchronizer不了解的可以参考我的博客:
[Java源码][并发J.U.C]---用代码一步步实现AQS(1)---独占锁的获取和释放
 [Java源码][并发J.U.C]---用代码一步步实现AQS(2)---独占锁中断式获取
 [Java源码][并发J.U.C]---用代码一步步实现AQS(3)---共享锁的获取和释放

framework.png

按照猜想,由于使用Condition对象con的前提是获得锁,那说明该线程不在AQS中的同步等待队列中,当该线程执行到await()时,就会加到到该con的条件等待队列并且会被阻塞,因为我们可以在ConditionObject中看到有firstWaiter和lastWaiter两个属性,就大概知道该con对象维护了一个链表. 所以当某个线程调用con.signal()方法时就唤醒该线程并让该线程尝试获得锁,之后的操作就会到获得锁的部分流程了,在AQS的分析中有详细分析,如果获得锁该线程就可以从con.await()中返回,如果没有获得锁就会进入AQS的同步等待队列中.

大体思路是如此, 接下来看看具体实现.具体细节还是需要通过代码中才可以了解清楚.

源码

先直接看await()方法

/**
         * Implements interruptible condition wait.
         * <ol>
         * <li> If current thread is interrupted, throw InterruptedException.
         * <li> Save lock state returned by {@link #getState}.
         * <li> Invoke {@link #release} with saved state as argument,
         *      throwing IllegalMonitorStateException if it fails.
         * <li> Block until signalled or interrupted.
         * <li> Reacquire by invoking specialized version of
         *      {@link #acquire} with saved state as argument.
         * <li> If interrupted while blocked in step 4, throw InterruptedException.
         * </ol>
         */
        public final void await() throws InterruptedException {
            if (Thread.interrupted()) // 如果当前线程被中断了 抛出异常
                throw new InterruptedException();
            Node node = addConditionWaiter();  // 添加一个新的节点到条件等待队列并返回该节点
            int savedState = fullyRelease(node); // 释放锁, 因为可能是重入锁,需要用fullRelease
            int interruptMode = 0;
            /**
             *  有两个条件退出while循环
             *  1. 当前节点出现在同步等待队列中
             *  2. 如果线程有中断(中断状态被改变)
             */
            while (!isOnSyncQueue(node)) {
                /**
                 *  线程休眠 有两种方式会从park方法退出
                 *  1. 被其他线程unpark唤醒
                 *  2. 线程发生中断
                 */
                LockSupport.park(this);
                // 如果线程有中断(中断状态被改变) 则break退出循环
                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                    break;
            }
System.out.println("in await() after while interrupteMode:" + interruptMode);
            // 尝试获得锁, 参数为之前释放锁的个数(在重入锁中可以理解为重入的次数)
            // acquireQueued(node, savedState) 返回true表示在获得锁的过程中线程有中断过
            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                interruptMode = REINTERRUPT;
            if (node.nextWaiter != null) // clean up if cancelled
                unlinkCancelledWaiters();
System.out.println("in await() final interrupteMode:" + interruptMode);
            if (interruptMode != 0)
                reportInterruptAfterWait(interruptMode);
        }

作用:
1. 如果当前线程中断状态是true 清除当前中断状态抛出中断异常InterruptedException.
2. 保存该锁的状态值, 在重入锁中表现为重入次数.
3. 释放锁
4. 阻塞当前线程除非被唤醒或者被中断
5. 以释放锁之前保存的状态值来重新尝试获得锁
6. 如果在4中是由于被中断才退出阻塞,该方法最终会抛出中断异常InterruptedException.

addConditionWaiter方法和

/**
         * 作用: 添加一个新的等待节点到条件等待队列(Condition wait queue)
         * @return 返回新生成的等待节点
         */
        private Node addConditionWaiter() {
            Node t = lastWaiter;
            // 如果最后一个等待节点不为空但是不为Condition(意味着被取消了)
            // 清除该节点并且重新设置最后一个等待节点
            if (t != null && t.waitStatus != Node.CONDITION) {
                unlinkCancelledWaiters();
                t = lastWaiter;
            }
            // 生成一个新的节点 节点类型是Node.CONDITION
            Node node = new Node(Thread.currentThread(), Node.CONDITION);
            if (t == null)
                firstWaiter = node;
            else
                t.nextWaiter = node;
            lastWaiter = node;
            return node;
        }

fullyRelease(node)方法
作用: 完全释放锁.

 /**
     * Invokes release with current state value; returns saved state.
     * Cancels node and throws exception on failure.
     * @param node the condition node for this wait
     * @return previous sync state 返回之前的同步状态
     */
    final int fullyRelease(Node node) {
        boolean failed = true;
        try {
            // 节点状态 持有锁的数量
            int savedState = getState();
            // 释放锁
            if (release(savedState)) {
                failed = false;
                return savedState;
            } else {
                // 如果释放锁失败 会抛出异常 必须是先获得锁 才可以调用wait
                throw new IllegalMonitorStateException();
            }
        } finally {
            // 如果失败的话修改节点的waitStatus为取消状态
            if (failed)
                node.waitStatus = Node.CANCELLED;
        }
    }

isOnSyncQueue方法

判断该node节点是否在AQS的同步队列中.

/**
     * 判断该节点是否转移到同步队列中
     * @param node the node
     * @return true if is reacquiring
     */
    final boolean isOnSyncQueue(Node node) {
        // 如果状态是CONDITION 或者 前驱节点为null 则表明肯定不在同步队列中 直接返回false.
        if (node.waitStatus == Node.CONDITION || node.prev == null)
            return false;
        // 此时node的状态不等于CONDITION并且前驱节点不为null
        // 如果node.next != null 则肯定在同步队列中
        if (node.next != null) // If has successor, it must be on queue
            return true;
        /*
         * 如果node的后驱节点是null,担心CAS失败,还是扫描确保
         */
        return findNodeFromTail(node);
    }

    /**
     * 从后往前扫描链表查找node是否在sync queue上.
     * @return true if present
     */
    private boolean findNodeFromTail(Node node) {
        Node t = tail;
        for (;;) {
            if (t == node)
                return true;
            if (t == null)
                return false;
            t = t.prev;
        }
    }

checkInterruptWhileWaiting(Node node)

判断当前线程在等待过程中是否有中断,关于REINTERRUPT和THROW_IE会在最后的例子2中用例子解释.

/** Mode meaning to reinterrupt on exit from wait */
        private static final int REINTERRUPT =  1;
        /** Mode meaning to throw InterruptedException on exit from wait */
        private static final int THROW_IE    = -1;

        /**
         * 检查中断
         * 返回THROW_IE       如果中断发生在被别的线程调用signal之前
         * 返回REINTERRUPT    如果中断发生在被别的线程调用signal之后
         * 返回0              如果没有发生中断
         */
        private int checkInterruptWhileWaiting(Node node) {
            return Thread.interrupted() ?
                    (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :
                    0;
        }
         /**
     * Transfers node, if necessary, to sync queue after a cancelled wait.
     * Returns true if thread was cancelled before being signalled.
     *
     * @param node the node
     * @return true if cancelled before the node was signalled
     */
    final boolean transferAfterCancelledWait(Node node) {
        if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {
            enq(node);
            return true;
        }
        /*
         * If we lost out to a signal(), then we can't proceed
         * until it finishes its enq().  Cancelling during an
         * incomplete transfer is both rare and transient, so just
         * spin.
         */
        while (!isOnSyncQueue(node))
            Thread.yield(); //线程让步
        return false;
    }

reportInterruptAfterWait

/**
         * Throws InterruptedException, reinterrupts current thread, or
         * does nothing, depending on mode.
         * 判断是否需要抛出异常
         */
        private void reportInterruptAfterWait(int interruptMode)
                throws InterruptedException {
            if (interruptMode == THROW_IE)
                throw new InterruptedException();
            else if (interruptMode == REINTERRUPT)
                selfInterrupt();
        }

唤醒方法signal

/**
         * 作用: 把条件等待队列中等待最长的线程所对应的节点(也就是头节点)转移到同步等待队列中
         * @throws IllegalMonitorStateException if {@link #isHeldExclusively}
         *         returns {@code false}
         */
        public final void signal() {
            if (!isHeldExclusively()) // 如果该线程没有获得锁则抛出异常
                throw new IllegalMonitorStateException();
            Node first = firstWaiter; // 获得条件等待队列中的第一个节点,也就是等待时间最长的节点
            if (first != null)
                doSignal(first); // 唤醒first节点
        }
/**
         * 作用: 把first节点从条件等待队列转移到同步等待队列,
         *       如果失败会尝试转移条件等待队列中的下一个节点,直到条件等待队列为空.
         * @param first
         */
        private void doSignal(Node first) {
            do {
                if ( (firstWaiter = first.nextWaiter) == null) //更新firstWaiter节点
                    lastWaiter = null;   // 当条件等待队列为空的时候更新lastWaiter节点
                first.nextWaiter = null; // 从条件等待队列中删除该节点
            } while (!transferForSignal(first) &&
                    (first = firstWaiter) != null);
            // 如果把first节点成功转移到同步等待队列中或者条件等待队列为空才会退出循环
        }
/**
     * 把一个条件等待队列的节点转移到同步队列
     * 返回ture如果成功, 失败的话说明该节点已经被取消
     * @param node the node
     * @return true if successfully transferred (else the node was
     * cancelled before signal)
     */
    final boolean transferForSignal(Node node) {
        /*
         * If cannot change waitStatus, the node has been cancelled.
         * 如果不能改变waitStatus,说明这个节点已经被取消了
         */
        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).
         * 如果前驱节点取消或尝试设置waitStatus失败，唤醒重新同步,(因为此时休眠可能会造成没有线程来唤醒)
         * 唤醒的过程中如果获得锁失败会调用shouldParkAfterFailedAcquire保证线程休眠后会有线程唤醒
         */
        Node p = enq(node); // 加入到同步等待队列并且返回该节点在同步等待队列中的前一个节点
        int ws = p.waitStatus;
        if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
            LockSupport.unpark(node.thread);
        return true;
    }

在transferForSignal方法中可以看到该节点在加入到同步队列后如果同步队列中该节点的前一个节点p的状态是取消状态或者设置成SIGNAL状态失败,则会通过唤醒线程的方式唤醒,否则会让前一个节点来唤醒它.(与await()方法呼应).

可以通过下面这个图来增加理解.

当前状态.png

signal.png

例子2

尝试在调用con.signal()方法前后分别中断线程thread并观察效果.

package com.sourcecode.reentrantreadwritelock;

import java.util.concurrent.TimeUnit;

public class TestConditionInterrupted {
    public static void main(String[] args) throws InterruptedException {
        Thread thread = new MyThread();
        thread.start();
        TimeUnit.SECONDS.sleep(5);
        lock.lock();
        System.out.println(getPrefix() + " gets lock");
        System.out.println(getPrefix() + "try to signal");
        //thread.interrupt();  //在signal前 await()会报异常 不能正常执行线程内的代码 interruptedMode=-1
        //Thread.sleep(10000);
        con.signal();
        thread.interrupt();   //在signal后 await()不报异常 正常执行线程内的代码 interruptedMode=1
        System.out.println(getPrefix() + "after signal");
        System.out.println(getPrefix() + " releases lock");
        lock.unlock();
    }

    static String getPrefix() {
        return Thread.currentThread().getName() + "==============";
    }

    static ReentrantLock lock = new ReentrantLock();
    static Condition con = lock.newCondition();

    static class MyThread extends Thread {
        public void run() {
            lock.lock();
            System.out.println(getPrefix() + "gets first lock");
            lock.lock();
            System.out.println(getPrefix() + "gets second lock");
            try {
                System.out.println(getPrefix() + "invoke await...");
                con.await();
                System.out.println(getPrefix() + "after await...");
                //TimeUnit.SECONDS.sleep(2);
            } catch (InterruptedException e) {
                e.printStackTrace();
            } finally {
                lock.unlock();
                System.out.println(getPrefix() + "release first lock");
                lock.unlock();
                System.out.println(getPrefix() + "release second lock");
                //lock.unlock();
                //System.out.println("release third lock");
            }
        }
    }
}

1. 在signal前 await()会报异常不能正常执行线程内的代码 interruptedMode=-1.
2. 在signal后 await()不报异常正常执行线程内的代码 interruptedMode=1.

总结

其他的几个方法都是类似的就不多说了.

参考

1. Java并发编程的艺术

[Java源码][并发J.U.C]---解析Condition

前言

例子1

实现思路

源码

addConditionWaiter方法和

isOnSyncQueue方法

checkInterruptWhileWaiting(Node node)

reportInterruptAfterWait

唤醒方法signal

例子2

总结

参考

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

程序员