队列同步器（AbstractQueuedSynchronizer）（二）

短命女 2022-02-26 00:52 264阅读 0赞

## 独占式同步状态获取与释放 ##

### 获取 ###

**    通过调用同步器的acquire(int arg)方法可以获取同步状态，该方法对中断不敏感，也就是由于线程获取同步状态失败后进入同步队列中，后续对线程进行中断操作时，线程不会从同 步队列中移出。**

public final void acquire(int arg) {
            if (!tryAcquire(arg) &&
                acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
                selfInterrupt();
        }

这段代码主要完成了同步状态获取、节点构造、加入同步队列以及在同步队列中自旋等待的相关工作，其主要逻辑是：

*  首先调用自定义同步器实现的tryAcquire(int arg)方法，该方法保证线程安全的获取同步状态。
 *  如果同步状态获取失败，则构造同步节点（独占式 Node.EXCLUSIVE，同一时刻只能有一个线程成功获取同步状态）并通过addWaiter(Node node) 方法将该节点加入到同步队列的尾部，最后调用acquireQueued(Node node,int arg)方法，使得该节点以“死循环”的方式获取同步状态。如果获取不到则阻塞节点中的线程，而被阻塞线程的唤醒主要依靠前驱节点的出队或阻塞线程被中断来实现。

下面看一下 addWaiter方法和end方法：

/**
         * 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
            Node pred = tail;//获取尾节点
            if (pred != null) {
                node.prev = pred;//设置当前节点的上一个节点为尾节点
                if (compareAndSetTail(pred, node)) {//将当前节点设为尾节点
                    pred.next = node;//前个节点的下个节点引向尾节点（双向队列）
                    return node;
                }
            }
            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) {
            for (;;) {
                Node t = tail;//设置为尾节点
                if (t == null) { // 如果为null则初始化
                    if (compareAndSetHead(new Node()))
                        tail = head;
                } else {//以下逻辑跟addWaiter相同
                    node.prev = t;
                    if (compareAndSetTail(t, node)) {
                        t.next = node;
                        return t;
                    }
                }
            }
        }

**    这两段代码通过使用compareAndSetTail(Node expect,Node update)方法来确保节点能够被线程安全添加。**

**    在enq(final Node node)方法中，同步器通过“死循环”来保证节点的正确添加，在“死循 环”中只有通过CAS将节点设置成为尾节点之后，当前线程才能从该方法返回，否则，当前线程不断地尝试设置。**

**    节点进入同步队列之后，就进入了一个自旋的过程，每个节点（或者说每个线程）都在自省地观察，当条件满足，获取到了同步状态，就可以从这个自旋过程中退出，否则依旧留在这个自旋过程中（并会阻塞节点的线程）。**

/**
         * 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();//如果上个节点为head节点，且获取锁成功
                    if (p == head && tryAcquire(arg)) {
                        setHead(node);//拿到资源后，将head指向该结点
                        p.next = null; //将head.next置为null方便GC回收以前的head结点
                        failed = false;
                        return interrupted;
                    }
                    if (shouldParkAfterFailedAcquire(p, node) &&
                        parkAndCheckInterrupt())
                        interrupted = true;//如果等待过程中被中断过，就将interrupted标记为true
                }
            } finally {
                if (failed)
                    cancelAcquire(node);
            }
        }

**  在acquireQueued(final Node node,int arg)方法中，当前线程在“死循环”中尝试获取同步状态，而只有前驱节点是头节点才能够尝试获取同步状态**，原因有两个：

1.  头节点是成功获取到同步状态的节点，而头节点的线程释放了同步状态之后，将会唤醒其后继节点，后继节点的线程被唤醒后需要检查自己的前驱节点是否是头节点。
2.  维护同步队列的FIFO原则。该方法中，节点自旋获取同步状态的行为下所示。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_16_color_FFFFFF_t_70][]

下面看一下如果失败他是如何处理的：

/**
         * 检查和更新未能获取的节点的状态。如果线程阻塞，返回true。
         *
         * @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) {
            int ws = pred.waitStatus;
            if (ws == Node.SIGNAL)//如果前驱节点状态值为SIGNAL，那当前线程应该被阻塞
                /*
                 * This node has already set status asking a release
                 * to signal it, so it can safely park.
                 */
                return true;
            if (ws > 0) {//CANCELLED状态，则跳过前驱节点，且重复查向前查找未被撤销的节点
                /*
                 * Predecessor was cancelled. Skip over predecessors and
                 * indicate retry.
                 */
                do {
                    node.prev = pred = pred.prev;
                } while (pred.waitStatus > 0);
              // 找到距离最近的非撤销前驱节点，并将它的下个节点设为当前节点
                pred.next = node;
            } else {
               //waitStatus值必为INITIAL或者PROPAGATE，
               //但当前节点需要一个SIGNAL信号，此时还不能阻塞线程
               //将前驱节点的状态值通过CAS操作更新为SIGNAL
                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);//调用park()使线程进入waiting状态
            return Thread.interrupted();//如果被唤醒，返回当前线程是否被中断
        }

这两步还有个重要作用就是：

**shouldParkAfterFailedAcquire和parkAndCheckInterrupt做的线程中断可以防止线程一直死循环抢占资源。**

**由于非首节点线程前驱节点出队或者被中断而从等待状态返回，随后检查自己的前驱是否是头节点，如果是则尝试获取同步状态。 可以看到节点和节点之间在循环检查的过程中基本不相互通信，而是简单地判断自己的前驱是否为头节点，这样就使得节点的释 放规则符合FIFO，并且也便于对过早通知的处理（过早通知是指前驱节点不是头节点的线程由于中断而被唤醒）。**

独占式同步状态获取流程，也就是acquire(int arg)方法调用流程：

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_16_color_FFFFFF_t_70 1][]

**    前驱节点为头节点，且能够获取同步状态的判断条件和线程进入等待状态，是获取同步状态的自旋过程。 当同步状态获取成功之后，当前线程从acquire(int arg)方法返回，表示获取了锁。**

### 释放 ###

**    通过调用同步器的release(int arg)方法可以释放同步状态，该方法在释放了同步状态之后，会唤醒其后继节点（进而使后继节点重新尝试获取同步状态）。**

/**
         * Releases in exclusive mode.  Implemented by unblocking one or
         * more threads if {@link #tryRelease} returns true.
         * This method can be used to implement method {@link Lock#unlock}.
         *
         * @param arg the release argument.  This value is conveyed to
         *        {@link #tryRelease} but is otherwise uninterpreted and
         *        can represent anything you like.
         * @return the value returned from {@link #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;
        }

**    该方法执行时，会唤醒头节点的后继节点线程，unparkSuccessor(Node node)方法使用 LockSupport来唤醒处于等待状态的线程。**

/**
         * Wakes up node's successor, if one exists.
         *
         * @param node the node
         */
        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)//将线程状态设为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);//唤醒
        }

**    总结：在获取同步状态时，同步器维护一个同步队列，获取状态失败的线程都会被加入到队列中并在队列中进行自旋；移出队列（或停止自旋）的条件是前驱节点为头节点且成功获取了同步状态。在释放同步状态时，同步器调用tryRelease(int arg)方法释放同步状态，然后唤醒头节点的后继节点。**

## 共享式同步状态获取与释放 ##

**     共享式获取与独占式获取最主要的区别在于同一时刻能否有多个线程同时获取到同步状态。**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_16_color_FFFFFF_t_70 2][]

### 获取 ###

通过调用同步器的acquireShared(int arg)方法可以共享式地获取同步状态：

/**
         * Acquires in shared mode, ignoring interrupts.  Implemented by
         * first invoking at least once {@link #tryAcquireShared},
         * returning on success.  Otherwise the thread is queued, possibly
         * repeatedly blocking and unblocking, invoking {@link
         * #tryAcquireShared} until success.
         *
         * @param arg the acquire argument.  This value is conveyed to
         *        {@link #tryAcquireShared} but is otherwise uninterpreted
         *        and can represent anything you like.
         */
        public final void acquireShared(int arg) {
            if (tryAcquireShared(arg) < 0)//尝试获取锁，需要自己实现
                doAcquireShared(arg);
        }
    
    
    
        /**
         * Acquires in shared uninterruptible mode.
         * @param arg the acquire argument
         */
        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;
                        }
                    }
                    //判断状态，寻找安全点，进入waiting状态，等着被unpark()或interrupt()
                    if (shouldParkAfterFailedAcquire(p, node) &&
                        parkAndCheckInterrupt())
                        interrupted = true;//当被中断设为1
                }
            } finally {
                if (failed)
                    cancelAcquire(node);
            }
        }
    
    
        /**
         * Sets head of queue, and checks if successor may be waiting
         * in shared mode, if so propagating if either propagate > 0 or
         * PROPAGATE status was set.
         *
         * @param node the node
         * @param propagate the return value from a tryAcquireShared
         */
        private void setHeadAndPropagate(Node node, int propagate) {
            Node h = head; // 获取未修改前的头节点
            setHead(node);//将当前节点设为头节点
            //如果还有剩余资源，继续唤醒下一个节点线程
            if (propagate > 0 || h == null || h.waitStatus < 0 ||
                (h = head) == null || h.waitStatus < 0) {
                Node s = node.next;
                if (s == null || s.isShared())
                    doReleaseShared();
            }
        }

** 在acquireShared(int arg)方法中，同步器调用tryAcquireShared(int arg)方法尝试获取同步状 态，tryAcquireShared(int arg)方法返回值为int类型，当返回值大于等于0时，表示能够获取到同步状态。因此，在共享式获取的自旋过程中，成功获取到同步状态并退出自旋的条件就是 tryAcquireShared(int arg)方法返回值大于等于0。**

**    在doAcquireShared(int arg)方法的自 旋过程中，如果当前节点的前驱为头节点时，尝试获取同步状态，如果返回值大于等于0，表示该次获取同步状态成功并从自旋过程中退出。**

### 释放 ###

共享式获取也需要释放同步状态，通过调用releaseShared(int arg)方法可以 释放同步状态：

/**
         * Releases in shared mode.  Implemented by unblocking one or more
         * threads if {@link #tryReleaseShared} returns true.
         *
         * @param arg the release argument.  This value is conveyed to
         *        {@link #tryReleaseShared} but is otherwise uninterpreted
         *        and can represent anything you like.
         * @return the value returned from {@link #tryReleaseShared}
         */
        public final boolean releaseShared(int arg) {
            if (tryReleaseShared(arg)) {//尝试释放资源
                doReleaseShared();//唤醒后继节点
                return true;
            }
            return false;
        }
    
    
    
     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;
            }
        }

**该方法在释放同步状态之后，将会唤醒后续处于等待状态的节点。对于能够支持多个线 程同时访问的并发组件（比如Semaphore），它和独占式主要区别在于tryReleaseShared(int arg) 方法必须确保同步状态（或者资源数）线程安全释放，一般是通过循环和CAS来保证的，因为释放同步状态的操作会同时来自多个线程。**

## 补充： ##

超时设置主要通过doAcquireNanos来实现：

/**
         * Acquires in exclusive timed mode.
         *
         * @param arg the acquire argument
         * @param nanosTimeout max wait time
         * @return {@code true} if acquired
         */
        private boolean doAcquireNanos(int arg, long nanosTimeout)
                throws InterruptedException {
            if (nanosTimeout <= 0L)
                return false;
            final long deadline = System.nanoTime() + nanosTimeout;//获得deadline时间
            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;
                    }
                    nanosTimeout = deadline - System.nanoTime();
             //如果deadline -当前时间小于等于0则直接退出循环
                    if (nanosTimeout <= 0L)
                        return false;
                    if (shouldParkAfterFailedAcquire(p, node) &&
                        nanosTimeout > spinForTimeoutThreshold)
                        LockSupport.parkNanos(this, nanosTimeout);
                    if (Thread.interrupted())
                        throw new InterruptedException();
                }
            } finally {
                if (failed)
                    cancelAcquire(node);
            }
        }

参考《Java并发编程的艺术》

CSDN连接：[https://download.csdn.net/download/qq\_37598011/11068576][https_download.csdn.net_download_qq_37598011_11068576]

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_16_color_FFFFFF_t_70]: /images/20220226/d969cf4e41e449b0b2738316f86b5e9a.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_16_color_FFFFFF_t_70 1]: /images/20220226/2b0e8dab86ba4c4397fa5370ab96a586.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3FxXzM3NTk4MDEx_size_16_color_FFFFFF_t_70 2]: /images/20220226/8b7ad965ca5a4b27ae8f8c56ddeb1dee.png
[https_download.csdn.net_download_qq_37598011_11068576]: https://download.csdn.net/download/qq_37598011/11068576