美文网首页javacode
AQS条件队列及中断机制

AQS条件队列及中断机制

作者: 九点半的马拉 | 来源:发表于2020-03-16 18:00 被阅读0次

    Condition条件队列

    当我们进行线程间的通信时,可以使用ReetrantLockCondition相结合,其中的await()signal()方法进行线程间的阻塞与唤醒。我将详细的解释其中的机制。
    ConditionObject
    ConditionObject是实现条件队列的关键,每个ConditionObject对象都维护一个单独的条件等待对列。一个AQS中可以有多个条件队列,但是只有一个同步队列。
    该博客的描述挺好的,有的部分是引用里面的。AQS解析与实战
    条件队列与同步队列的联系


    1) 调用了await()方法的线程,会被加入到ConditionObject等待队列中,并且唤醒同步队列中head节点的下一个节点
    2) 线程在某个ConditionObject对象上调用signal()方法后,等待队列中的firstWaiter会被加入到同步队列中,等待被唤醒。
    3)当线程调用unLock()方法解锁后,同步队列中的head节点的下一个节点会被唤醒。
    public class ConditionObject implements Condition, java.io.Serializable {
            private transient Node firstWaiter;
            private transient Node lastWaiter;
    

    每个ConditionObject对应一个条件队列,它记录该队列的头节点和尾节点。
    条件队列是单向的,而同步队列是双向的,会有前驱指针。

    await()方法

    public final void await() throws InterruptedException {
                if (Thread.interrupted())
                    throw new InterruptedException();
                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;
                if (node.nextWaiter != null) // clean up if cancelled
                    unlinkCancelledWaiters();
                if (interruptMode != 0)
                    reportInterruptAfterWait(interruptMode);
            }
    
    private Node addConditionWaiter() {
                Node t = lastWaiter;
                // If lastWaiter is cancelled, clean out.
                if (t != null && t.waitStatus != Node.CONDITION) {
                    unlinkCancelledWaiters();
                    t = lastWaiter;
                }
                Node node = new Node(Thread.currentThread(), Node.CONDITION);
                if (t == null)
                    firstWaiter = node;
                else
                    t.nextWaiter = node;
                lastWaiter = node;
                return node;
            }
    

    整体的流程如下:

    • 执行await()时,会新创建一个节点并放入到该条件队列尾部。
    • 然后释放锁,并唤醒同步队列中的Head节点的后一个节点。
    • 然后while循环,将该节点阻塞,直达该节点被放入到同步节点或者被中断了,才退出循环。
    • 退出循环后,开始调用acquireQueued()不断尝试拿锁。
    • 拿到锁后,会清空条件队列中被取消的节点。
    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);
            }
    
    final boolean transferForSignal(Node node) {
            if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
                return false;
            Node p = enq(node);
            int ws = p.waitStatus;
            if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
                LockSupport.unpark(node.thread);
            return true;
        }
    
    private Node enq(final Node node) {
            for (;;) {
                Node t = tail;
                if (t == null) { // Must initialize
                    if (compareAndSetHead(new Node()))
                        tail = head;
                } else {
                    node.prev = t;
                    if (compareAndSetTail(t, node)) {
                        t.next = node;
                        return t;
                    }
                }
            }
        }
    

    主要过程:
    将条件队列中的头节点放入到同步队列尾部,并获取它在同步队列中的前驱节点
    如果他前驱节点的状态是取消状态或者设置设置前驱节点为Signal失败,则唤醒当前节点,
    唤醒后会执行在await()方法阻塞后的代码,会进行不断尝试获取锁。

    线程的中断

    当我们想执行线程中断时,一般会调用interrupt方法,对于初学者可能以为线程会立即中断执行,但是如果线程处于运行状态,不会受影响,会继续执行下去。为什么会出现这种情况呢?

    java的线程中断机制,主要是依靠中断标志位实现的,当调用interrupt方法时,会将中断标志位设置为ture,然后针对线程此时的不同线程状态,会有不同的处理。具体的处理情况可以阅读这篇文章: java中断详细介绍及其对各种线程状态的影响分析

    执行中断后,当线程处于waiting状态会抛出中断异常并将清空中断标志位。
    前面写了一篇AQS的文章,介绍了acqure()方法,最后一步调用了selfInterrupt(),这是当前线程进行中断。一般我们加锁会调动lock()方法,但是实际上还有一个方法lockInterruptibly()。我将通过ReetrantLock类介绍它俩的区别。

    final void lock() {
                if (compareAndSetState(0, 1))
                    setExclusiveOwnerThread(Thread.currentThread());
                else
                    acquire(1);
    }
    public final void acquire(int arg) {
            if (!tryAcquire(arg) &&
                acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
                selfInterrupt();
        }
    static void selfInterrupt() {
            Thread.currentThread().interrupt();
        }
    private final boolean parkAndCheckInterrupt() {
            LockSupport.park(this);
            return Thread.interrupted();
        }
    

    首先分析下lock()方法,首先如果state为0,那么通过CAS机制将state设置为1,然后将当前线程设置为锁的拥有者;其他情况则调用acquire(1)尝试加锁,具体的尝试加锁机制就不介绍了,在之前文章写过,它是一个自旋操作,如果获取失败,会将线程park挂起,进入阻塞状态,当其他线程执行完,会唤醒该线程,执行向下执行,这时会执行Thread.interrupted(),会判断当前线程是否被中断,如果该线程拿到锁,会执行selfInterrupt()方法,将当前线程进行中断,但是之后自己设计的逻辑中没有如果没有处理中断的代码,那么它会一直执行下去,或者将当前线程设置为waitiing状态,就会抛出中断异常,并消除中断标志。

    public void lockInterruptibly() throws InterruptedException {
            sync.acquireInterruptibly(1);
        }
        
    public final void acquireInterruptibly(int arg)
                throws InterruptedException {
            if (Thread.interrupted())
                throw new InterruptedException();
            if (!tryAcquire(arg))
                doAcquireInterruptibly(arg);
        }
        
    private void doAcquireInterruptibly(int arg)
            throws InterruptedException {
            final Node node = addWaiter(Node.EXCLUSIVE);
            boolean failed = true;
            try {
                for (;;) {
                    final Node p = node.predecessor();
                    if (p == head && tryAcquire(arg)) {
                        setHead(node);
                        p.next = null; // help GC
                        failed = false;
                        return;
                    }
                    if (shouldParkAfterFailedAcquire(p, node) &&
                        parkAndCheckInterrupt())
                        throw new InterruptedException();
                }
            } finally {
                if (failed)
                    cancelAcquire(node);
            }
        }
    

    现在再介绍一下lockInterruptibly的机制。
    首先它会判断当前线程是否中断,如果有,就马上抛出异常;没有中断,则开始尝试获得锁,获取失败,将该节点放入到同步队列中,并执行自旋操作:
    1)如果该节点的前驱节点头节点,并且该节点获取到了锁,则设置当前节点是头节点,并返回
    2)其他情况,就需要将它的前置节点设置为Siginal,并将当前节点所属的线程挂起,当重新唤醒后,如果发生中断,会直接抛出异常,并将该节点从同步队列中删除

    相关文章

      网友评论

        本文标题:AQS条件队列及中断机制

        本文链接:https://www.haomeiwen.com/subject/dbbnehtx.html