AQS深入理解 hasQueuedPredecessors源码分析 JDK8

约定不等于承诺〃 2023-03-14 10:41 7阅读 0赞

### 文章目录 ###

*  前言
 *  流程
 *  hasQueuedPredecessors分析
 *  为什么先读取tail，再读取head
 *   *  先读取tail，再读取head的好处
     *  是否需要考虑 指令重排序
 *  虚假返回的true和false
 *   *  虚假返回true
     *  虚假返回false

# 前言 #

> Queries whether any threads have been waiting to acquire longer than the current thread.

首先要知道hasQueuedPredecessors这个方法是为 公平锁为设计的函数，看名字就知道，这是用来判断有没有别的线程排在了当前线程的前面。

[JUC框架 系列文章目录][JUC_]

# 流程 #

已经熟悉流程的同学可跳过。  
首先看到ReentrantLock的内部类FairSync：

public void lock() { 
            sync.lock();  //此时sync成员是一个FairSync实例
        }
    
        static final class FairSync extends Sync { 
            private static final long serialVersionUID = -3000897897090466540L;
    
            final void lock() { 
                acquire(1);
            }
    
            /** * 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();
                int c = getState();
                if (c == 0) { 
                    if (!hasQueuedPredecessors() &&  //调用在这里
                        compareAndSetState(0, acquires)) { 
                        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;
            }
        }

我们调用ReentrantLock的lock方法时，最终会调用到FairSync实例的acquire方法，而acquire方法已经在AQS中有实现了：

public final void acquire(int arg) { 
            if (!tryAcquire(arg) &&  //tryAcquire是子类的实现，见上面代码
                acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
                selfInterrupt();
        }
    
        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);
            }
        }

现在观察tryAcquire的逻辑（直接考虑当前同步器的state为0）：

*  发现如果调用`hasQueuedPredecessors`返回true的话，后继代码都不走了，直接就返回false了。
 *  如果调用`hasQueuedPredecessors`返回false的话，才去尝试CAS修改同步器的状态：
    
     *  如果CAS修改同步器的状态 成功，通过setExclusiveOwnerThread设置 独占模式下的线程。返回true。
     *  如果CAS修改同步器的状态 失败。返回false。

可见`hasQueuedPredecessors`的返回值很重要，直接影响到 是否会去 尝试CAS修改同步器的state。

如果只有一个线程来调用ReentrantLock的lock方法，且该同步器state还是0，那么此时tryAcquire（看名字就知道，尝试获得锁）已经执行成功并返回true。再观察`!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)`，可见后续操作也不会执行了，if分支也不会进入了。否则执行后续操作`acquireQueued(addWaiter(Node.EXCLUSIVE), arg)`。

`addWaiter`简单地说，就是把当前线程包装成一个node放在队尾，并返回这个刚建的node。`acquireQueued`简单的说，就是执行一个for循环，每次循环都会`tryAcquire`一下，如果tryAcquire成功，设置头节点后结束循环；如果tryAcquire失败，调用shouldParkAfterFailedAcquire设置上一个node的signal信号，然后调用parkAndCheckInterrupt便阻塞在里面。重点在于，这个for循环不是连续执行的，而可能 不停地 阻塞然后被唤醒 ，重复着这样的过程。

# hasQueuedPredecessors分析 #

public final boolean hasQueuedPredecessors() { 
            // The correctness of this depends on head being initialized
            // before tail and on head.next being accurate if the current
            // thread is first in queue.
            Node t = tail; // Read fields in reverse initialization order
            Node h = head;
            Node s;
            return h != t &&
                ((s = h.next) == null || s.thread != Thread.currentThread());
        }

要分析上面return的逻辑，必须要看线程获取同步器state失败时（`tryAcquire(arg)`返回false），要执行的入队操作`addWaiter`。

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;//这里只是执行一个快速操作，它和enq里的else分支的逻辑一样
                if (compareAndSetTail(pred, node)) { 
                    pred.next = node;
                    return node;
                }
            }
            enq(node);//如果上面快速操作没有成功，再执行enq
            return node;
        }
    
        private Node enq(final Node node) { 
            for (;;) { //使用for循环，保证入队成功
                Node t = tail;
                if (t == null) {  // 第一次入队，没有dummy node的存在，需先创建它
                    if (compareAndSetHead(new Node()))
                        tail = head;
                } else {  // 至少有一个node，尝试入队
                    node.prev = t;
                    if (compareAndSetTail(t, node)) { 
                        t.next = node;
                        return t;
                    }
                }
            }
        }

首先解释下，hasQueuedPredecessors在先后读取完tail和head后，如果这二者只有一个为null（另一个不为null），那么只可能出现“head不为null，tail为null”的情况：

*  从`if (compareAndSetHead(new Node()))`到`tail = head;`的间隙可知，除非一个线程恰好在`tail = head;`之前读取了tali域（`Node t = tail; Node t = head;`），那么才可能发生 此时 head不为null，tail为null的情况。
 *  否则，其他情况下。head和tail要都为null，要么都不为null。

接下来分析`hasQueuedPredecessors`的返回判断。首先要知道，`hasQueuedPredecessors`返回true代表有别的线程在CHL队列中排了当前线程之前；返回false代表当前线程处于CHL队列的第一个线程。

1.  分析`h != t`返回false的情况。此时`hasQueuedPredecessors`返回false。
    
     *  当h和t都为null，返回false。此时说明队列为空，还从来没有Node入过队。
     *  当h和t都指向同一个Node，也返回false。此时说明队列中只有一个dummy node，那说明没有线程在队列中。
2.  分析`h != t`返回true，且`(s = h.next) == null`返回true，直接短路后面。此时`hasQueuedPredecessors`返回true。
    
     *  既然`h != t`返回true，说明h和t不相等，先考虑特殊情况（上面讲到的出现“head不为null，tail为null”的情况，此时head是空node，next成员肯定为null），那么说明有一个线程正在执行`enq`，且它正好执行到`if (compareAndSetHead(new Node()))`到`tail = head;`的间隙。但这个线程肯定不是当前线程，所以不用判断后面短路的`s.thread != Thread.currentThread()`了，因为当前线程连`enq`都没开始执行，但另一个线程都开始执行`enq`了，那不就是说明当前线程排在别人后面了，别的线程马上就要入队了。
     *  既然`h != t`返回true，说明h和t不相等，再考虑二者都不为null。那此时队列中已经至少有一个等待中的线程了，那说明当前线程肯定排在别人后面了。
3.  分析`h != t`返回true，且`(s = h.next) == null`返回false，且`s.thread != Thread.currentThread()`返回true。此时`hasQueuedPredecessors`返回true。如果`s.thread != Thread.currentThread()`返回false。此时`hasQueuedPredecessors`返回false。
    
     *  现在知道head不为null，而且head.next也不为null了（`(s = h.next) == null`返回false）。我们也知道队列中第一个等待的线程存放在head.next里（注意，head为dummy node，不存放线程），那么如果head.next的线程不是当前线程，那即说明当前线程已经排在别人线程后面了。

# 为什么先读取tail，再读取head #

前面说到，hasQueuedPredecessors在先后读取完tail和head后，如果这二者只有一个为null（另一个不为null），那么只可能出现“head不为null，tail为null”的情况。

现在假设当前线程正要执行这两句：

Node t = tail; 
            Node h = head;

另一个线程则正要执行这两句（标记了1 2）：

private Node enq(final Node node) { 
            for (;;) { 
                Node t = tail;
                if (t == null) {  
                    if (compareAndSetHead(new Node())) // 1
                        tail = head; // 2
                } else { 
                    node.prev = t;
                    if (compareAndSetTail(t, node)) { 
                        t.next = node;
                        return t;
                    }
                }
            }
        }

现在任意排列这四条语句的顺序，但保持 各个线程的先后顺序。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FubGlhbjUyMw_size_16_color_FFFFFF_t_70]  
可见只会出现上面三种情况。

但如果hasQueuedPredecessors先读取head后读取tail，则可能发生“head为null，tail不为null”的情况（分析类似上图），而接下来的`(s = h.next) == null`这里，就抛出空指针异常了。

## 先读取tail，再读取head的好处 ##

当`Node t = tail; Node h = head;`执行完毕，只可能出现三种情况：

*  `h`和`t`都为null
 *  `h`和`t`都不为null
 *  `h`不为null，但`t`为null

不可能出现：

*  `t`不为null，但`h`为null

从而避免了`(s = h.next) == null`这里的空指针异常。

## 是否需要考虑 指令重排序 ##

如果考虑指令重排序，那么可能 就变成了 先读取head再读取tail 了，那就可能不对了。但由于这两句都是volatile写操作，每个volatile写操作前面加StoreStore屏障，后面加StoreLoad内存屏障。

> StoreStore 屏障：保证在 volatile 写之前，其前面的所有普通写操作，都已经**刷新到主内存中**。  
> StoreLoad 屏障：避免 volatile 写，与后面可能有的 volatile 读 / 写操作**重排序**。

根据StoreLoad屏障的作用，我们可以保证线程是 先读取tail后读取head。

# 虚假返回的true和false #

## 虚假返回true ##

> Note that because cancellations due to interrupts and timeouts may occur at any time, a true return does not guarantee that some other thread will acquire before the current thread.

注意，由于中断和超时导致的取消可能随时发生，因此返回true不能保证某些其他线程将在当前线程之前获取。

比如当你调用了`acquireInterruptibly`时，如果因为获取不到锁，会暂时阻塞在`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);//②
            }
        }

*  现在，阻塞在`parkAndCheckInterrupt`里的线程已经被唤醒，但在此之前被设置了中断状态，所以马上抛出InterruptedException。然后再执行cancelAcquire，被唤醒的线程的所在的node才会被移除出队列。
 *  但在cancelAcquire执行完之前，head.next则还是一个即将被移除的node。而此时 别的线程执行hasQueuedPredecessors肯定返回true，所以说它是虚假的true（考虑队列只有dummy node和一个即将被移除的node）。

## 虚假返回false ##

> Likewise, it is possible for another thread to win a race to enqueue after this method has returned false, due to the queue being empty.

同样，由于队列为空，此方法返回false后，另一个线程也有可能赢得竞争。

这个情况比较容易想到，当队列为空时（连dummy node都没有），同时有两个线程正在执行`tryAcquire`，且两个线程都刚执行完了`hasQueuedPredecessors`。

*  此时两个线程获得的hasQueuedPredecessors的返回值肯定都是false。
 *  但接下来的CAS操作`compareAndSetState(0, acquires)`，就不一定是谁胜出了。
 *  失败了那一方，就认为 之前调用hasQueuedPredecessors返回的false，是一个虚假的false。

protected final boolean tryAcquire(int acquires) { 
                final Thread current = Thread.currentThread();
                int c = getState();
                if (c == 0) { 
                    if (!hasQueuedPredecessors() && //两个线程都执行到这里
                        compareAndSetState(0, acquires)) { 
                        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;
            }

当然针对这种虚假的false，是不会有什么坏影响的，因为`acquireQueued`中会再次执行`tryAcquire`的。

[JUC_]: https://blog.csdn.net/anlian523/article/details/106598910
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FubGlhbjUyMw_size_16_color_FFFFFF_t_70]: /images/20230312/e11d63bbab6c423dae81796f91ca5902.png