zoukankan      html  css  js  c++  java
  • 谈论高并发(二十二)解决java.util.concurrent各种组件(四) 深入了解AQS(二)

    上一页介绍AQS其基本设计思路以及两个内部类Node和ConditionObject实现 聊聊高并发(二十一)解析java.util.concurrent各个组件(三) 深入理解AQS(一) 这篇说一说AQS的主要方法的实现。AQS和CLHLock的最大差别是,CLHLock是自旋锁,而AQS使用Unsafe的park操作让线程进入等待(堵塞)。


    线程增加同步队列,和CLHLock一样,从队尾入队列,使用CAS+轮询的方式实现无锁化。

    入队列后设置节点的prev和next引用,形成双向链表的结构

    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;
                    }
                }
            }
        }

    线程指定独享还是共享方式增加队列,先尝试增加一次,假设失败再用enq()轮询地尝试,比方addWaiter(Node.EXCLUSIVE), addWaiter(Node.SHARED)

    private Node addWaiter(Node mode) {
            Node node = new Node(Thread.currentThread(), mode);
            // Try the fast path of enq; backup to full enq on failure
            Node pred = tail;
            if (pred != null) {
                node.prev = pred;
                if (compareAndSetTail(pred, node)) {
                    pred.next = node;
                    return node;
                }
            }
            enq(node);
            return node;
        }

    唤醒后继节点,最典型的情况就是在线程释放锁后,会唤醒后继节点。会从节点的next開始,找到一个后继节点,假设next是null。就从队尾開始往head找,直到找到最靠近当前节点的兴许节点。 waitStatus <= 0的隐含意思是线程没有被取消。 然后用LockSupport唤醒这个找到的后继节点的线程。

    这种方法类似于CLHLock里面释放锁时,通知兴许节点来获取锁。AQS使用了堵塞的方式,所以这种方法的兴许方法是acquireXXX方法,它负责将兴许节点唤醒,兴许节点再依据状态去推断是否获得锁

    private void unparkSuccessor(Node node) {
            /*
             * If status is negative (i.e., possibly needing signal) try
             * to clear in anticipation of signalling.  It is OK if this
             * fails or if status is changed by waiting thread.
             */
            int ws = node.waitStatus;
            if (ws < 0)
                compareAndSetWaitStatus(node, ws, 0);
    
            /*
             * Thread to unpark is held in successor, which is normally
             * just the next node.  But if cancelled or apparently null,
             * traverse backwards from tail to find the actual
             * non-cancelled successor.
             */
            Node s = node.next;
            if (s == null || s.waitStatus > 0) {
                s = null;
                for (Node t = tail; t != null && t != node; t = t.prev)
                    if (t.waitStatus <= 0)
                        s = t;
            }
            if (s != null)
                LockSupport.unpark(s.thread);
        }

    共享模式下的释放操作。从队首開始向队尾扩散,假设节点的waitStatu是SIGNAL,就唤醒后继节点。假设waitStatus是0,就设置标记成PROPAGATE

    private void doReleaseShared() {
            for (;;) {
                Node h = head;
                if (h != null && h != tail) {
                    int ws = h.waitStatus;
                    if (ws == Node.SIGNAL) {
                        if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                            continue;            // loop to recheck cases
                        unparkSuccessor(h);
                    }
                    else if (ws == 0 &&
                             !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                        continue;                // loop on failed CAS
                }
                if (h == head)                   // loop if head changed
                    break;
            }
        }

    取消获取操作,要把节点从同步队列中去除。通过链表操作将它的前置节点的next指向它的后继节点集合。假设该节点是在队尾,直接删除就可以,否则要通知后继节点去获取锁

    private void cancelAcquire(Node node) {
            // Ignore if node doesn't exist
            if (node == null)
                return;
    
            node.thread = null;
    
            // Skip cancelled predecessors
            Node pred = node.prev;
            while (pred.waitStatus > 0)
                node.prev = pred = pred.prev;
    
            // predNext is the apparent node to unsplice. CASes below will
            // fail if not, in which case, we lost race vs another cancel
            // or signal, so no further action is necessary.
            Node predNext = pred.next;
    
            // Can use unconditional write instead of CAS here.
            // After this atomic step, other Nodes can skip past us.
            // Before, we are free of interference from other threads.
            node.waitStatus = Node.CANCELLED;
    
            // If we are the tail, remove ourselves.
            if (node == tail && compareAndSetTail(node, pred)) {
                compareAndSetNext(pred, predNext, null);
            } else {
                // If successor needs signal, try to set pred's next-link
                // so it will get one. Otherwise wake it up to propagate.
                int ws;
                if (pred != head &&
                    ((ws = pred.waitStatus) == Node.SIGNAL ||
                     (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&
                    pred.thread != null) {
                    Node next = node.next;
                    if (next != null && next.waitStatus <= 0)
                        compareAndSetNext(pred, predNext, next);
                } else {
                    unparkSuccessor(node);
                }
    
                node.next = node; // help GC
            }
        }

    独占模式而且不可中断地获取队列锁的操作。这种方法在ConditionObject.await()中被使用。当线程被Unsafe.unpark唤醒后,须要调用acquireQueued来获取锁,从而结束await(). accquireQueued()方法要么获得锁,要么被tryAcquire方法抛出的异常打断,假设抛出异常,最后在finally里面取消获取

    值得注意的是仅仅有节点的前驱节点是head的时候,才干获得锁。

    这里隐含了一个意思,就是head指向当前获得锁的节点。当程序进入if(p == head and tryAcquire(arg))这个分支时。表示线程获得了锁或者被中断。将自己设置为head,将next设置为null.

    shouldParkAfterFailedAcquired()方法的目的是将节点的前驱节点的waitStatus设置为SIGNAL,表示会通知兴许节点,这样兴许节点才干放心去park。而不用操心被丢失唤醒的通知。

    parkAndCheckInterupt()方法会真正运行堵塞,并返回中断状态,这种方法有两种情况返回。一种是park被unpark唤醒。这时候中断状态为false。还有一种情况是park被中断了,因为这个accquireQueued方法是不可中断的版本号。所以即使线程被中断了,也仅仅是设置了中断标志为true,没有跑出中断异常。

    在支持中断的获取版本号里。这时会抛出中断异常。

    这种方法能够理解为Lock的lock里没有获取锁的分支。在CLHLock自旋锁的实现里。是对前驱节点的状态自旋,而AQS是堵塞。所以这里是在同步队列里面进入了堵塞状态。等待被前驱节点释放锁时唤醒。

    释放锁时会依据状态调用unparkSuccessor()方法来唤醒兴许节点,这样就会在这种方法里面把堵塞的线程唤醒并获得锁。

    队列锁的优点是线程都在多个共享状态上自旋或堵塞。所以unparkSuccessor()方法仅仅会唤醒它后继没有取消的节点。

    而取消仅仅有两种情况,中断或者超时

    final boolean acquireQueued(final Node node, int arg) {
            boolean failed = true;
            try {
                boolean interrupted = false;
                for (;;) {
                    final Node p = node.predecessor();
                    if (p == head && tryAcquire(arg)) {
                        setHead(node);
                        p.next = null; // help GC
                        failed = false;
                        return interrupted;
                    }
                    if (shouldParkAfterFailedAcquire(p, node) &&
                        parkAndCheckInterrupt())
                        interrupted = true;
                }
            } finally {
                if (failed)
                    cancelAcquire(node);
            }
        }

    独占模式支持中断的获取队列锁操作。能够看到和不支持中断版本号的差别,这里假设parkAndCheckInterrupt()方法返回时显示被中断了,就抛出中断异常


    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);
            }
        }

    独占模式限时获取队列锁操作, 这个获取的总体逻辑和前面的类似,差别是它支持限时操作,假设等待时间大于spinForTimeoutThreshold,就使用堵塞的方式等待,否则用自旋等待。使用了LockSupport.parkNanos()方法来实现限时地等待,并支持中断

    这里隐含的一个含义是parkNanos方法退出有3种方式,

    1. 限时到了自己主动退出,这时候会超时

    2. 没有到限时被唤醒了。这时候是不超时的

    3. 被中断

    private boolean doAcquireNanos(int arg, long nanosTimeout)
            throws InterruptedException {
            long lastTime = System.nanoTime();
            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 true;
                    }
                    if (nanosTimeout <= 0)
                        return false;
                    if (shouldParkAfterFailedAcquire(p, node) &&
                        nanosTimeout > spinForTimeoutThreshold)
                        LockSupport.parkNanos(this, nanosTimeout);
                    long now = System.nanoTime();
                    nanosTimeout -= now - lastTime;
                    lastTime = now;
                    if (Thread.interrupted())
                        throw new InterruptedException();
                }
            } finally {
                if (failed)
                    cancelAcquire(node);
            }
        }

    共享模式获得队列锁操作。获得操作也是从head的下一个节点開始,和独占模式仅仅unparkSuccessor一个节点不同,共享模式下,等head的兴许节点被唤醒了,它要扩散这样的共享的获取,使用setHeadAndPropagate操作,把自己设置为head,而且把释放的状态往下传递。这里採用了链式唤醒的方法,1个节点负责唤醒1个兴许节点,直到不能唤醒。当后继节点是共享模式isShared,就调用doReleaseShared来唤醒后继节点

    doReleaseShared会从head開始往后检查状态,假设节点是SIGNAL状态,就唤醒它的后继节点。

    假设是0就标记为PROPAGATE, 等它释放锁的时候会再次唤醒后继节点。

    这里有个隐含的意思:

    1. 增加同步队列并堵塞的节点,它的前驱节点仅仅会是SIGNAL。表示前驱节点释放锁时。后继节点会被唤醒。shouldParkAfterFailedAcquire()方法保证了这点,假设前驱节点不是SIGNAL,它会把它改动成SIGNAL。

    这里不是SIGNAL就有可能是PROPAGATE

    2. 造成前驱节点是PROPAGATE的情况是前驱节点获得锁时。会唤醒一次后继节点。但这时候后继节点还没有增加到同步队列,所以临时把节点状态设置为PROPAGATE,当后继节点增加同步队列后,会把PROPAGATE设置为SIGNAL,这样前驱节点释放锁时会再次doReleaseShared,这时候它的状态已经是SIGNAL了,就能够唤醒兴许节点了


    private void doAcquireShared(int arg) {
            final Node node = addWaiter(Node.SHARED);
            boolean failed = true;
            try {
                boolean interrupted = false;
                for (;;) {
                    final Node p = node.predecessor();
                    if (p == head) {
                        int r = tryAcquireShared(arg);
                        if (r >= 0) {
                            setHeadAndPropagate(node, r);
                            p.next = null; // help GC
                            if (interrupted)
                                selfInterrupt();
                            failed = false;
                            return;
                        }
                    }
                    if (shouldParkAfterFailedAcquire(p, node) &&
                        parkAndCheckInterrupt())
                        interrupted = true;
                }
            } finally {
                if (failed)
                    cancelAcquire(node);
            }
        }
    
    private void setHeadAndPropagate(Node node, int propagate) {
            Node h = head; // Record old head for check below
            setHead(node);
            /*
             * Try to signal next queued node if:
             *   Propagation was indicated by caller,
             *     or was recorded (as h.waitStatus) by a previous operation
             *     (note: this uses sign-check of waitStatus because
             *      PROPAGATE status may transition to SIGNAL.)
             * and
             *   The next node is waiting in shared mode,
             *     or we don't know, because it appears null
             *
             * The conservatism in both of these checks may cause
             * unnecessary wake-ups, but only when there are multiple
             * racing acquires/releases, so most need signals now or soon
             * anyway.
             */
            if (propagate > 0 || h == null || h.waitStatus < 0) {
                Node s = node.next;
                if (s == null || s.isShared())
                    doReleaseShared();
            }
        }
    
    private void doReleaseShared() {
            for (;;) {
                Node h = head;
                if (h != null && h != tail) {
                    int ws = h.waitStatus;
                    if (ws == Node.SIGNAL) {
                        if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                            continue;            // loop to recheck cases
                        unparkSuccessor(h);
                    }
                    else if (ws == 0 &&
                             !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                        continue;                // loop on failed CAS
                }
                if (h == head)                   // loop if head changed
                    break;
            }
        }
    

    tryXXXX 方法,这几个方法是给子类重写的。用来扩展响应的同步器操作

    protected boolean tryAcquire(int arg) {
            throw new UnsupportedOperationException();
        }
    
    protected boolean tryRelease(int arg) {
            throw new UnsupportedOperationException();
        }
    
    protected int tryAcquireShared(int arg) {
            throw new UnsupportedOperationException();
        }
    
    protected boolean tryReleaseShared(int arg) {
            throw new UnsupportedOperationException();
        }
    

    独占模式获取操作的顶层方法,假设没有tryAcquired,或者没有获得队列锁,就中断
    public final void acquire(int arg) {
            if (!tryAcquire(arg) &&
                acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
                selfInterrupt();
        }

    独占模式释放操作的顶层方法,假设tryRelease()成功,那么就唤醒后继节点去获取锁
    public final boolean release(int arg) {
            if (tryRelease(arg)) {
                Node h = head;
                if (h != null && h.waitStatus != 0)
                    unparkSuccessor(h);
                return true;
            }
            return false;
        }


  • 相关阅读:
    C# 窗体间传值方法大汇总(转)
    STM32 配置PC13~PC15
    STM32的USART发送数据时如何使用TXE和TC标志
    STM32_NVIC寄存器详解
    protel99se 问题汇总(不定期更新)
    STM32串口IAP实验笔记
    Keil MDK下如何设置非零初始化变量(复位后变量值不丢失)
    STM32定时器配置(TIM1-TIM8)高级定时器+普通定时器,定时计数模式下总结
    帮助类-AD域操作
    GitHub贡献第一的公司是谁?微软开源软件列表
  • 原文地址:https://www.cnblogs.com/lcchuguo/p/5036172.html
Copyright © 2011-2022 走看看