jdk10-ReentrantLock重入锁源码分析（下）

作者: 洋葱520 | 来源:发表于2018-08-06 15:13 被阅读0次

https://www.jianshu.com/p/d560c2d9ea8e
上一篇对ReentrantLock的lock和unlock方法做了详细的讲解这篇将接着讲解Condition，

1. condition 生产者消费者列子

condition 是依赖于 ReentrantLock 的，不管是调用 await 进入等待还是 signal 唤醒，都必须获取到锁才能进行操作。
每个 ReentrantLock 实例可以通过调用多次 newCondition 产生多个 ConditionObject 的实例：

public class ReentrantLock implements Lock, java.io.Serializable {
    public Condition newCondition() {
        return sync.newCondition();
    }
}

public class ConditionTest {
    final Lock lock = new ReentrantLock();
    final Condition producer = lock.newCondition();
    final Condition consumer = lock.newCondition();

    final int max = 100;
    final LinkedList<Integer> item = new LinkedList<>();

    // 生产
    public void put() throws InterruptedException {
        lock.lock();
        try {
            while (item.size() == max) {
                producer.await(); // 队列已满，生产者等待
            }
            item.add(new Random().nextInt());
            consumer.signal(); // 生产成功，通知消费者
        } finally {
            lock.unlock();
        }
    }

    // 消费
    public void take() throws InterruptedException {
        lock.lock();
        try {
            while (item.size() == 0) {
                consumer.await(); // 队列为空，消费者等待
            }
            System.out.println(item.poll());
            producer.signal(); // 被我消费掉一个，给生产者发个通知
        } finally {
            lock.unlock();
        }
    }
}

2. AQS.ConditionObject

我们可以看到ConditionObject中利用两个属性来组成条件队列

public class ConditionObject implements Condition, java.io.Serializable { 
        /** First node of condition queue. 条件队列第一个结点 */
        private transient Node firstWaiter;
        /** Last node of condition queue.  条件队列最后一个节点*/
        private transient Node lastWaiter;

        /** 代表 await 返回的时候，需要重新设置中断状态 */
        private static final int REINTERRUPT =  1;
        /** 代表 await 返回的时候，需要抛出 InterruptedException 异常 */
        private static final int THROW_IE    = -1;
}

3. 线程1调用 ConditionObject.await()

执行方法链

1、ConditionObject.await()，线程1在获取到锁后，调用await()

包装线程为node加入到条件等待队列

释放锁

线程1节点不在AQS同步等待队列中，阻塞当前线程，
（signal会把线程移到AQS队列中，就跳出死循环）

线程1先执行到此结束

ConditionObject::
        public final void await() throws InterruptedException {
            if (Thread.interrupted()) // 中断抛出异常退出
                throw new InterruptedException();
1、包装当前线程node加入到条件队列
            Node node = addConditionWaiter();
2、释放锁，返回释放之前的状态值（这个方法锁释放）
            int savedState = fullyRelease(node);
            int interruptMode = 0;
3、如果不是在同步队列中，一直while
            while (!isOnSyncQueue(node)) {
4、阻塞当前线程
                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);
        }

2、 ConditionObject.addConditionWaiter(),此方法是在还没释放锁的情况下，调用所以线程安全

获取条件队列最后一个节点，如果最后一个节点是null，头结点加入，不是null ，t.nextWaiter=node，重置lastWaiter

获取最后一个节点 t，如果 t == null，头结点加入队列，firstWaiter指向node

如果 t != null ,t.nextWaiter 指向node

从新使lastWaiter 指向当前线程node

ConditionObject ::

        private Node addConditionWaiter() {
1、ReentrantLock.isHeldExclusively() 是否是当前线程持有锁
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            Node t = lastWaiter;
2、 如果最后节点是取消状态，清理条件队列，（因为此次是线程1所以为null）
            if (t != null && t.waitStatus != Node.CONDITION) {
                unlinkCancelledWaiters();
                t = lastWaiter;
            }
1、包装当前线程为node
            Node node = new Node(Node.CONDITION);
2、加入到条件队列
            if (t == null)
                firstWaiter = node;
            else
                t.nextWaiter = node;
3、last指向最新节点
            lastWaiter = node;
            return node;
        }

        protected final boolean isHeldExclusively() {
            return getExclusiveOwnerThread() == Thread.currentThread();
        }

线程执行后队列情况

3、 AQS.fullyRelease(node) 完全释放锁

获取当前线程的状态

调用AQS.release(int) 这个方法就是ReentrantLock.unlock()调用的那个

释放成功返回释放之前的值

AQS :: 
    final long fullyRelease(Node node) {
        try {
            long savedState = getState();
            if (release(savedState))
                return savedState;
            throw new IllegalMonitorStateException();
        } catch (Throwable t) {
            node.waitStatus = Node.CANCELLED;
            throw t;
        }
    }

4、 AQS.isOnSyncQueue() 如果不在AQS同步等待队列中返回false

第一个判断，状态是condition ：= 条件队列中，和前驱节点不存在：=在条件队列中

只有同步等待队列 next才不会为null
可以通过判断 node.prev() != null 判断node 在阻塞队列吗？不能
AQS 的入队方法，首先设置的是 node.prev 指向 tail，然后cas可能失败

经过上面两步的判断可以明显的知道节点在AQS的末尾，
从后到前遍历同步等待队列查看是否在队列中

AQS ::
    final boolean isOnSyncQueue(Node node) {
1、状态是condition，和前驱节点不存在，那么节点不会再同步等待队列中
        if (node.waitStatus == Node.CONDITION || node.prev == null)
            return false;
2、只有同步等待队列 next才不会为null
        if (node.next != null) // If has successor, it must be on queue
            return true;
        return findNodeFromTail(node);
    }
    private boolean findNodeFromTail(Node node) {
        for (Node p = tail;;) {
            if (p == node)
                return true;
            if (p == null)
                return false;
            p = p.prev;
        }
    }

4. ConditionObject.signal() 唤醒线程1

移动第一个线程(等待最长时间的)到AQS等待队列中，并且获取锁

获取第一个线程节点，非null，调用doSignal(first)

        public final void signal() {
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            Node first = firstWaiter;
            if (first != null)
                doSignal(first);
        }

1、doSignal(first)

首先重置 first.nextWaiter = null; 把此节点 清理出条件队列

transferForSignal把节点移到AQS队列返回true 退出循环

假如移动失败（节点被中断已经移动过了），继续循环移动下一个

        private void doSignal(Node first) {
            do {
                if ( (firstWaiter = first.nextWaiter) == null)
                    lastWaiter = null;
1、 清理节点
                first.nextWaiter = null;
            } while (  !transferForSignal(first) &&  (first = firstWaiter) != null  );
        }

    final boolean transferForSignal(Node node) {
        /*
         * 如果失败，说明线程可能已经中断
         */
        if (!node.compareAndSetWaitStatus(Node.CONDITION, 0))
            return false;

        Node p = enq(node);
        int ws = p.waitStatus;
1、ws > 0 说明 node 在阻塞队列中的前驱节点取消了等待锁，直接唤醒 node 对应的线程
2、如果 ws <= 0, 那么 compareAndSetWaitStatus 将会被调用，上篇介绍的时候说过，节点入队后，需要把前驱节点的状态设为 Node.SIGNAL(-1)
        if (ws > 0 || !p.compareAndSetWaitStatus(ws, Node.SIGNAL))
            LockSupport.unpark(node.thread);
        return true;
    }

上面介绍了正常的 await()，signal()过程，下面我们分析，特殊情况

5. 线程1调用 await() 被其他线程中断执行流程

中断执行流程

1、ConditionObject.await()，线程1在获取到锁后，调用await()

包装线程为node加入到条件等待队列

释放锁

线程1节点不在AQS同步等待队列中，阻塞当前线程，
（signal会把线程移到AQS队列中，就跳出死循环）

中断后的执行流程：

中断 --> 重置中断状态，加入到AQS等待队列 interruptMode = THROW_IE，跳出循环 注意没有清理节点

acquireQueued获取到锁

unlinkCancelledWaiters上面《中断》并不会把自己清理出条件队列，所以清理取消的节点

根据interruptMode 判断是否抛出异常还是设置中断状态（第4步清理掉了中断状态）

    public final void await() throws InterruptedException {
            if (Thread.interrupted()) // 中断抛出异常退出
                throw new InterruptedException();
1、包装当前线程node加入到条件队列
            Node node = addConditionWaiter();
2、释放锁，返回释放之前的状态值（这个方法锁释放）
            int savedState = fullyRelease(node);
            int interruptMode = 0;
3、如果不是在同步队列中，一直while
            while (!isOnSyncQueue(node)) {
4、阻塞当前线程
                LockSupport.park(this);
5、中断判断，跳出循环，加入到AQS等待队列  interruptMode  = THROW_IE
                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                    break;
            }
6. interruptMode   = THROW_IE 所以继续走，acquireQueued获取到锁
            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                interruptMode = REINTERRUPT;
7. 上面中断并不会把自己清理出条件队列，所以清理取消的节点
            if (node.nextWaiter != null) // clean up if cancelled
                unlinkCancelledWaiters();
8. 根据interruptMode 判断是否抛出异常还是设置中断状态
            if (interruptMode != 0)
                reportInterruptAfterWait(interruptMode);
        }

2、checkInterruptWhileWaiting(node)

没有中断：return 0，signalled之前中断：THROW_IE ，之后中断：REINTERRUPT

特别注意 node.compareAndSetWaitStatus(Node.CONDITION, 0），失败表示signal已经移动节点，这样防止并发

        /**
         * 没有中断：0，signalled之前中断：THROW_IE ，之后中断：REINTERRUPT 
         */
        private int checkInterruptWhileWaiting(Node node) {
            return Thread.interrupted() ?
                (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :
                0;
        }

    final boolean transferAfterCancelledWait(Node node) {
        if (node.compareAndSetWaitStatus(Node.CONDITION, 0)) {
            enq(node);
            return true;
        }
        /*
         * 到这里是因为 CAS 失败，肯定是因为 signal 方法已经将 waitStatus 设置为了 0
         * signal 方法会将节点转移到阻塞队列，但是可能还没完成，这边自旋等待其完成
         * 当然，这种事情还是比较少的吧：signal 调用之后，没完成转移之前，发生了中断
         */
        while (!isOnSyncQueue(node))
            Thread.yield();
        return false;
    }

3、acquireQueued(node, savedState) 获取锁

这个就不说了，上篇有讲解

4、unlinkCancelledWaiters();

等待队列是一个单向链表，遍历链表将已经取消等待的节点清除出去

Node t ：作为当前需要清理的线程节点

Node trail ：作为Node t 的上一个节点，链表的上一个节点

循环 t=next;

如果 t 状态不是在条件队列中

清理 t 指向下一个节点的引用

把 t 节点的上一个节点的next指向t的next，

        private void unlinkCancelledWaiters() {
            Node t = firstWaiter;
            Node trail = null;
            while (t != null) {
                Node next = t.nextWaiter; 
                1、t = 线程1，如果状态不等于CONDITION，需要清理
                if (t.waitStatus != Node.CONDITION) {
                    2、清理 t 的 next引用
                    t.nextWaiter = null;
                    3、清理 t 的上一个节点，指向t的引用
                    if (trail == null)
                        firstWaiter = next;
                    else
                        trail.nextWaiter = next;
                    4、next=null 说明遍历完毕，把lastWaiter指向t的上一个节点
                    if (next == null)
                        lastWaiter = trail;
                } else {
                    trail = t;                
                }

                5、每次while循环把t重置指向他的next节点
                t = next;
            }
        }

5、reportInterruptAfterWait(interruptMode);

判断是抛出异常，还是设置中断状态

        private void reportInterruptAfterWait(int interruptMode)
            throws InterruptedException {
            if (interruptMode == THROW_IE)
                throw new InterruptedException();
            else if (interruptMode == REINTERRUPT)
                selfInterrupt();
        }

    static void selfInterrupt() {
        Thread.currentThread().interrupt();
    }

6.假如线程中断的时候，又执行了唤醒 transferForSignal 移动节点

我们可以看到在入队之前，都先使用cas，保证了节点不会重复加入AQS队列

我上文中也有特别注意，表明：中断和signal同时发生的时候，处理方式

    final boolean transferForSignal(Node node) {
        /*
         * 如果失败，说明线程可能已经中断
         */
        if (!node.compareAndSetWaitStatus(Node.CONDITION, 0))
            return false;
    }

    final boolean transferAfterCancelledWait(Node node) {
        if (node.compareAndSetWaitStatus(Node.CONDITION, 0)) {
            enq(node);
            return true;
    }

7. signalAll

可以很简单的看到do while循环移动节点

        public final void signalAll() {
            if (!isHeldExclusively())
                throw new IllegalMonitorStateException();
            Node first = firstWaiter;
            if (first != null)
                doSignalAll(first);
        }
        private void doSignalAll(Node first) {
            lastWaiter = firstWaiter = null;
            do {
                Node next = first.nextWaiter;
                first.nextWaiter = null;
                transferForSignal(first);
                first = next;
            } while (first != null);
        }
    final boolean transferForSignal(Node node) {
        if (!node.compareAndSetWaitStatus(Node.CONDITION, 0))
            return false;

        Node p = enq(node);
        int ws = p.waitStatus;
        if (ws > 0 || !p.compareAndSetWaitStatus(ws, Node.SIGNAL))
            LockSupport.unpark(node.thread);
        return true;
    }

8. awaitUninterruptibly() 不抛出异常 InterruptedException

加入队列，阻塞线程，获取锁，不过多介绍

        public final void awaitUninterruptibly() {
            Node node = addConditionWaiter();
            int savedState = fullyRelease(node);
            boolean interrupted = false;
            while (!isOnSyncQueue(node)) {
                LockSupport.park(this);
                if (Thread.interrupted())
                    interrupted = true;
            }
            if (acquireQueued(node, savedState) || interrupted)
                selfInterrupt();
        }

9. 具备超时的await

三个方法都是和上面的await差不多，就是多个时间判断

public final long awaitNanos(long nanosTimeout) throws InterruptedException

public final boolean awaitUntil(Date deadline) throws InterruptedException

public final boolean await(long time, TimeUnit unit) throws InterruptedException

public final long awaitNanos(long nanosTimeout)  throws InterruptedException {
            if (Thread.interrupted())
                throw new InterruptedException();
            // We don't check for nanosTimeout <= 0L here, to allow
            // awaitNanos(0) as a way to "yield the lock".
1、过期时间=当前时间 + 等待时长
            final long deadline = System.nanoTime() + nanosTimeout;
            long initialNanos = nanosTimeout;

            Node node = addConditionWaiter();
            int savedState = fullyRelease(node);
            int interruptMode = 0;

            while (!isOnSyncQueue(node)) {
2、时间到了
                if (nanosTimeout <= 0L) {
3、取消等待 transferAfterCancelledWait(node) 移动节点到 AQS等待队列
                    transferAfterCancelledWait(node);
                    break;
                }
4、如果等待时间大于 自旋阈值  ，阻塞
                if (nanosTimeout > SPIN_FOR_TIMEOUT_THRESHOLD)
                    LockSupport.parkNanos(this, nanosTimeout);
5、检测中断
                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                    break;
6、等待时间 = 过期时间-当前时间
                nanosTimeout = deadline - System.nanoTime();
            }
            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                interruptMode = REINTERRUPT;
            if (node.nextWaiter != null)
                unlinkCancelledWaiters();
            if (interruptMode != 0)
                reportInterruptAfterWait(interruptMode);
7、剩余时间
            long remaining = deadline - System.nanoTime(); // avoid overflow
            return (remaining <= initialNanos) ? remaining : Long.MIN_VALUE;
        }

        public final boolean await(long time, TimeUnit unit)
                throws InterruptedException {
            long nanosTimeout = unit.toNanos(time);
            if (Thread.interrupted())
                throw new InterruptedException();
            // We don't check for nanosTimeout <= 0L here, to allow
            // await(0, unit) as a way to "yield the lock".
            final long deadline = System.nanoTime() + nanosTimeout;
            Node node = addConditionWaiter();
            int savedState = fullyRelease(node);
            boolean timedout = false;
            int interruptMode = 0;
            while (!isOnSyncQueue(node)) {
                if (nanosTimeout <= 0L) {
                    timedout = transferAfterCancelledWait(node);
                    break;
                }
                if (nanosTimeout > SPIN_FOR_TIMEOUT_THRESHOLD)
                    LockSupport.parkNanos(this, nanosTimeout);
                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                    break;
                nanosTimeout = deadline - System.nanoTime();
            }
            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                interruptMode = REINTERRUPT;
            if (node.nextWaiter != null)
                unlinkCancelledWaiters();
            if (interruptMode != 0)
                reportInterruptAfterWait(interruptMode);
            return !timedout;
        }

        public final boolean awaitUntil(Date deadline)
                throws InterruptedException {
            long abstime = deadline.getTime();
            if (Thread.interrupted())
                throw new InterruptedException();
            Node node = addConditionWaiter();
            int savedState = fullyRelease(node);
            boolean timedout = false;
            int interruptMode = 0;
            while (!isOnSyncQueue(node)) {
                if (System.currentTimeMillis() >= abstime) {
                    timedout = transferAfterCancelledWait(node);
                    break;
                }
                LockSupport.parkUntil(this, abstime);
                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
                    break;
            }
            if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
                interruptMode = REINTERRUPT;
            if (node.nextWaiter != null)
                unlinkCancelledWaiters();
            if (interruptMode != 0)
                reportInterruptAfterWait(interruptMode);
            return !timedout;
        }

jdk10-ReentrantLock重入锁源码分析（下）

1. condition 生产者消费者列子

2. AQS.ConditionObject

3. 线程1调用 ConditionObject.await()

4. ConditionObject.signal() 唤醒线程1

5. 线程1调用 await() 被其他线程中断执行流程

6.假如线程中断的时候，又执行了唤醒 transferForSignal 移动节点

7. signalAll

8. awaitUninterruptibly() 不抛出异常 InterruptedException

9. 具备超时的await

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

jdk10-ReentrantLock重入锁源码分析（下）

1. condition 生产者消费者列子

2. AQS.ConditionObject

3. 线程1调用 ConditionObject.await()

4. ConditionObject.signal() 唤醒线程1

5. 线程1调用 await() 被其他线程中断执行流程

6.假如线程中断的时候 ，又执行了唤醒 transferForSignal 移动节点

7. signalAll

8. awaitUninterruptibly() 不抛出异常 InterruptedException

9. 具备超时的await

相关文章

网友评论

延伸阅读

深度阅读

栏目导航

热点阅读

6.假如线程中断的时候，又执行了唤醒 transferForSignal 移动节点