【AQS】独占锁的获取

扫码关注公众号:Java 技术驿站

发送:vip
将链接复制到本浏览器,永久解锁本站全部文章

【公众号:Java 技术驿站】 【加作者微信交流技术,拉技术群】
免费领取10G资料包与项目实战视频资料

文章首发于:clawhub.club


AQS中独占锁的获取由acquire(int arg)方法实现:

     /**
         * Acquires in exclusive mode, ignoring interrupts.  Implemented
         * by invoking at least once {@link #tryAcquire},
         * returning on success.  Otherwise the thread is queued, possibly
         * repeatedly blocking and unblocking, invoking {@link
         * #tryAcquire} until success.  This method can be used
         * to implement method {@link Lock#lock}.
         * 描述了独占的方式获取锁的流程,忽略获取锁的过程中的中断,
         *
         * @param arg the acquire argument.  This value is conveyed to
         *            {@link #tryAcquire} but is otherwise uninterpreted and
         *            can represent anything you like.
         */
        public final void acquire(int arg) {
            if (!tryAcquire(arg) &&
                    acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
                selfInterrupt();
        }

先看一下获取锁的流程图加深些了解:

2019102910020\_1.png

acquire.png

这个方法中调用了四个方法:

  1. tryAcquire(int arg)
    由子类实现,获取锁。
  2. addWaiter(Node mode)
    获取锁失败后,将等待线程封装成Node加入等待队列,由AQS实现。
  3. acquireQueued(final Node node, int arg)
    在队列中如果其前驱节点是头节点,就循环获取锁,获取锁成功就返回。
    如果其前驱不是头节点,或者是头节点但是获取锁失败,挂起当前线程。由AQS实现。
  4. selfInterrupt()
    自我中断,当获取锁的时候,发生中断时记录下来,推迟到抢锁结束后中断线程。

下面根据源码,详细的看看这几个方法,tryAcquire(int arg)参考了ReentrantLock中公平锁FairSync 。、

tryAcquire(int arg)

AQS中这个方法时protected且返回UnsupportedOperationExeception的,即表明此方法由子类实现。

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

ReentrantLock的内部类FairSync继承自Sync,而Sync继承自AbstractQueuedSynchronizer。
tryAcquire(int acquires)源码翻译:

     /**
             * Fair version of tryAcquire.  Don't grant access unless
             * recursive call or no waiters or is first.
             */
            protected final boolean tryAcquire(int acquires) {
                //获取当前线程
                final Thread current = Thread.currentThread();
                //获取当前锁的状态,即AQS中的state属性
                int c = getState();
                //c==0即锁没有被占用
                if (c == 0) {

                    //首先检查自己是不是处于头节点的后继节点,即队列中有没有排在我前面的节点
                    //之后将state设置为1
                    if (!hasQueuedPredecessors() &&
                            compareAndSetState(0, acquires)) {
                        //上面两步都成功之后,将AbstractOwnableSynchronizer的exclusiveOwnerThread设置为当前线程,就此获取锁成功。
                        setExclusiveOwnerThread(current);
                        //获取锁成功
                        return true;
                    }
                }
                //如果当前的线程持有这个锁
                else if (current == getExclusiveOwnerThread()) {
                    //记录锁重入次数
                    int nextc = c + acquires;
                    if (nextc < 0)
                        throw new Error("Maximum lock count exceeded");
                    setState(nextc);
                    return true;
                }
                //获取锁失败
                return false;
            }
        }

从源码中可以看出,有两个地方修改了state状态,一个是CAS,一个是普通的set方法,只有当前线程获取锁的时候,才可以直接用set方法。

addWaiter(Node mode)

从addWaiter(Node.EXCLUSIVE)可以看出,将当前线程封装成了独占模式,

    /**
         * Creates and enqueues node for current thread and given mode.
         * 根据给定的模式(独占或者共享)创建一个队列节点。
         *
         * @param mode Node.EXCLUSIVE for exclusive, Node.SHARED for shared
         * @return the new node
         */
        private Node addWaiter(Node mode) {
            Node node = new Node(Thread.currentThread(), mode);
            // Try the fast path of enq; backup to full enq on failure
            //尝试快速路径的enq;故障时备份到完整的enq
            Node pred = tail;
            //队尾不为空,CAS尝试一次入队
            if (pred != null) {
                node.prev = pred;
                if (compareAndSetTail(pred, node)) {
                    pred.next = node;
                    return node;
                }
            }
            //队尾为空,或者上一次尝试入队失败,进入enq方法
            enq(node);
            //返回新插入的节点
            return node;
        }
       /**
         * Inserts node into queue, initializing if necessary. See picture above.
         * 将节点插入队列,必要时初始化。
         *
         * @param node the node to insert
         * @return node's predecessor
         */
        private Node enq(final Node node) {
            //循环CAS,即最终必定会将node节点插入到同步队列中
            for (; ; ) {
                //获取队尾
                Node t = tail;
                //如果队尾为null,则初始化新节点
                if (t == null) { // Must initialize
                    //队首设置,CAS操作,只有一个线程会成功,队首只是一个标志位,后面才存储真正的等待线程节点
                    if (compareAndSetHead(new Node()))
                        //队尾指向队首
                        tail = head;
                    //继续循环
                } else {
                    //队尾不为空才能进入此代码块
                    //当前节点的前驱指向队尾
                    node.prev = t;
                    //CAS,设置队尾,有可能此时队尾已经改变,那就继续循环,如果设置成功,则当前节点变为队尾
                    if (compareAndSetTail(t, node)) {
                        //老的队尾的后继指向当前节点(新队尾),
                        // 如果在此之前,从前往后遍历等待队列,可能会获取不到新插入的新节点,
                        // 如果是从后往前遍历,就不会出现问题。
                        t.next = node;
                        //返回老的尾节点,即新插入节点的前驱节点
                        return t;
                    }
                }
            }
        }

从上述代码中可以看出,独占模式节点中nextWaiter为null,并且头节点一定是空节点。

acquireQueued(final Node node, int arg)

这个方法相对来说比较复杂,直接看源码的注释吧:

     /**
         * Acquires in exclusive uninterruptible mode for thread already in
         * queue. Used by condition wait methods as well as acquire.
         * 获取队列中已存在线程的独占不可中断模式。
         * 用于条件等待方法以及获取。
         * 能走到这一步,那么这个等待锁的线程所封装的节点一定在等待队列中
         *
         * @param node the node
         * @param arg  the acquire argument
         * @return {@code true} if interrupted while waiting
         */
        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);
                        // help GC
                        p.next = null;
                        failed = false;
                        return interrupted;
                    }
                    //走到这,要么节点的前驱不是头节点,要么是获取锁失败了。
                    //如果前驱节点waitStatus为SIGNAL,挂起当前线程,并且检查中断
                    //如果前驱节点waitStatus不为SIGNAL,最终将其设置为SIGNAL
                    if (shouldParkAfterFailedAcquire(p, node) &&
                            parkAndCheckInterrupt())
                        //在这之前,线程已经被挂起了,坐等解阻塞
                        interrupted = true;
                }
            } finally {
                if (failed)
                    cancelAcquire(node);
            }
        }
     /**
         * Checks and updates status for a node that failed to acquire.
         * Returns true if thread should block. This is the main signal
         * control in all acquire loops.  Requires that pred == node.prev.
         *
         * @param pred node's predecessor holding status
         * @param node the node
         * @return {@code true} if thread should block
         */
        private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
            // 获得前驱节点的ws
            int ws = pred.waitStatus;
            //如果前驱节点是SIGNAL,返回true,就会执行挂起当前线程操作
            if (ws == Node.SIGNAL)
                /*
                 * This node has already set status asking a release
                 * to signal it, so it can safely park.
                 */
                return true;
            //CANCELLED=1,所以如果ws>0,表示前驱节点已经释放锁了
            if (ws > 0) {
                /*
                 * Predecessor was cancelled. Skip over predecessors and
                 * indicate retry.
                 */
                do {
                    //1,前驱节点指向前驱节点的前驱
                    //2,当前节点的前驱指向前驱节点
                    // 即跳过取消了等待锁的前驱节点
                    node.prev = pred = pred.prev;
                } while (pred.waitStatus > 0);
                //新前驱节点的后继指向当前节点
                pred.next = node;
            } else {
                //这种情况就直接将前驱节点的ws设置为SIGNAL
                /*
                 * waitStatus must be 0 or PROPAGATE.  Indicate that we
                 * need a signal, but don't park yet.  Caller will need to
                 * retry to make sure it cannot acquire before parking.
                 */
                compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
            }
            return false;
        }
     /**
         * Convenience method to park and then check if interrupted
         *
         * @return {@code true} if interrupted
         */
        private final boolean parkAndCheckInterrupt() {
            //线程被挂起了,不会向下执行了,等待被唤醒
            LockSupport.park(this);
            return Thread.interrupted();
        }

能执行到这个方法,一定是等待队列中的节点,先尝试获取锁,获取失败之后再来判断是否将当前线程挂起。

下次分析独占锁的释放的时候,会有释放后继节点的操作。


来源:https://www.jianshu.com/p/347ea7f881f8

赞(0) 打赏
版权归原创作者所有,任何形式的转载请联系博主:daming_90:Java 技术驿站 » 【AQS】独占锁的获取

评论 抢沙发

  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

微信扫一扫打赏