【JUC源码】JUC核心：AQS（二）同步队列源码分析（独占锁）

小鱼儿 2022-10-31 01:42 188阅读 0赞

> AQS 系列：
> 
>  *  [【JUC源码】JUC核心：AQS（一）底层结构分析][JUC_JUC_AQS]
>  *  [【JUC源码】JUC核心：AQS（二）同步队列源码分析（独占锁）][JUC_JUC_AQS 1]
>  *  [【JUC源码】JUC核心：AQS（三）同步队列源码分析（共享锁）][JUC_JUC_AQS 2]
>  *  [【JUC源码】JUC核心：AQS（四）条件队列源码分析][JUC_JUC_AQS 3]
>  *  [【JUC源码】JUC核心：关于AQS的几个问题][JUC_JUC_AQS 4]

同步队列：

*  作用：管理多个线程的休眠与唤醒
 *  策略：可以执行的线程 = RUNNABLE 状态 && tryAcquire() 成功
    
     *  独占模式（EXCLUSIVE）：队首持锁，唤醒队二后 tryAcquire() 尝试拿锁，队三及以后休眠
     *  共享模式（SHARED）：相较于独占模式只唤醒队二 ，共享模式还唤醒所有 mode=shared 节点（多了一步）  
        注：这里需要明确一点，独占和共享是**对于加锁而言**（能否多线程同时获锁），释放锁时没有独占和共享的概念
 *  状态
    
     *  初始化（0）：入队的初始状态
     *  SIGINAL（-1）：若当前 node 后面还有 node，就要从 0->SIGNAL
     *  CANCELLED（1）：拿锁失败或出现异常，会在置为 CANCELLED 后删除，但并发时可能会暂时维持

## 1.独占-加锁 ##

### acquire() ###

该方法用作获取锁。排他模式下，acquire 方法由子类的 lock 方法直接调用。如下图是 Reentrantlock 的静态内部类 Sync 和 NonfairSync：

![在这里插入图片描述][20200920155545794.png_pic_center]

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center]  
注：从图中也可以看出 NonfairSync 的 lock 方法是**非公平**的，因为当前线程直接就有获取锁的机会（CAS修改state成功），不是必须要进入同步队列，接受同步器的调度。

*  尝试获得锁，成功直接放回
 *  失败，加入同步队列，等待拿锁

public final void acquire(int arg) {
        
        // tryAcquire 方法是需要子类去实现的
        // CAS修改state判断能否拿到锁,拿到锁return true，不会再进入addWaiter
        if (!tryAcquire(arg) &&
            // addWaiter 入参代表是排他模式
            acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
            selfInterrupt();
    }

> PS：这里一定要明白，AQS 中并没有实现 tryAcquire() 方法，它是交由子类去实现的，因为它是使用 AQS 加锁的关键。

如下图，是 Reentrantlock中 NonfairSync 的 tryAcquire 方法的具体实现

![在这里插入图片描述][20200920163515366.png_pic_center]

// Reentrantlock.sync#nonfairTryAcquire()
    final boolean nonfairTryAcquire(int acquires) {
        
    	// 获取当前线程
        final Thread current = Thread.currentThread(); 
        // 获取当前同步器的状态
        int c = getState(); 
        // 若c=0，表示没有线程持锁，即有机会获取锁
        if (c == 0) {
         
        	// 尝试将state通过CAS设置为1 
            if (compareAndSetState(0, acquires)) {
         
            	// 若CAS成功，表示当前线程可以拿锁，则将拿锁线程设为当前
                setExclusiveOwnerThread(current); 
                return true; // 返回true
            }
        }
        // 如果当前线程已经获得锁了
        else if (current == getExclusiveOwnerThread()) {
         
        	 // 重入，+1
            int nextc = c + acquires;
            if (nextc < 0) // overflow
                throw new Error("Maximum lock count exceeded");
            setState(nextc); 
            // 返回true
            return true; 
        }
    
    	// 否则，返回false，表示获取不到锁
        return false; 
    }

### addWaiter() ###

回到 acquire() ，可以看到 tryAcquire() 后，就是 addWaiter()，将等待的线程加入同步队列。

> PS：这里注意一点，对于同步队列节点的所有操作，都要是线程安全的，即通过 CAS

private Node addWaiter(Node mode) {
        
        // 创建并初始化 Node
        Node node = new Node(Thread.currentThread(), mode);
        // 在自旋前，先尝试看能否直接加到队尾，若成功直接返回
        // 这种做法大部分都可以一次成功，节省了自旋的开销
        Node pred = tail;
        // 当前同步队列不能为空，因为为空时还要设置head
        if (pred != null) {
        
    		// 将新节点node的前置节点设置为tail（双向链表）
            node.prev = pred; 
            // 为了保证线程安全，CAS交换尾结点，然后再连接上一个tail
            if (compareAndSetTail(pred, node)) {
        
                pred.next = node;
                return node;
            }
        }
        // 当直接加入队列失败（即CAS失败或者队列为空），需要通过自旋保证node加入到队尾
        enq(node);
        return node;
    }

### enq() ###

*  线程加入同步队列中方法，追加到队尾
 *  这里需要重点注意的是，返回值是添加 node 的前一个节点

private Node enq(final Node node) {
        
        // 自旋，保证在出现竞争时也能安全加入同步队列
        for (;;) {
        
            // 得到队尾节点
            Node t = tail;
            // 如果队尾为空，说明当前同步队列都没有初始化
            // tail = head = new Node();
            if (t == null) {
        
            	// 则新建一个空 node 作为头
                if (compareAndSetHead(new Node()))
                    tail = head;
            // 队尾不为空，将当前节点追加到队尾
            } else {
        
                node.prev = t;
                // node 追加到队尾
                if (compareAndSetTail(t, node)) {
        
                    t.next = node;
                    return t;
                }
            }
        }

### acquireQueued（核心方法) ###

管理同步队列（拿锁+休眠）

*  队二尝试拿锁，成功后取代队首
    
     *  同步队列队首是已经获得锁的节点
     *  队二有两种情况：1.新node进入直接就是队二 2.队n被唤醒后发现此时他已经前进了到了队二
     *  若队二拿锁失败（非公平锁），则（继续）进入休眠；但是它仍是队二，队首也没删
 *  阻塞队3-队n
    
     *  前提是自旋使自己前一个节点的状态变成 signal（SIGNAL代表后面一定有待唤醒，不会被遗忘）
     *  等线程一个个醒来后，再尝试拿锁

final boolean acquireQueued(final Node node, int arg) {
        
        boolean failed = true;
        try {
        
            boolean interrupted = false;
            // 自旋
            for (;;) {
        
                // predecessor是Node中的方法，作用是放回当前节点的上一个节点（prev）
                final Node p = node.predecessor();
     
                // 队二是有资格夺锁（tryAcquire）的节点
                // 若成功则把自己设置为队首，失败就（再）进入休眠
                if (p == head && tryAcquire(arg)) {
         // 如果当前节点是队二，且可以获取到锁
                    // 将当前node置为head，实际上就是删除已经释放锁的节点
                    setHead(node);
                    // p（之前获得锁的节点）被回收，next置为null是为了help gc
                    p.next = null; 
                    failed = false;
                    return interrupted;
                }
    
                // shouldParkAfterFailedAcquire 检验node能否休眠（pre=SIGNAL)，若不能则设置pre=SIGNAL
                // parkAndCheckInterrupt 阻塞当前线程
                if (shouldParkAfterFailedAcquire(p, node) &&
                    // 线程是在这个方法里面阻塞的，醒来的时候仍然在无限 for 循环里面，就能再次自旋尝试获得锁
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
        
            // 如果获得node的锁失败或异常，将 node 置为 CANCELLED，并删除
            // 因此，在并发情况下，同步队列中的任何位置节点都可能短暂CANCELLED，但最后一定会被删掉
            if (failed)
                cancelAcquire(node);
        }
    }

### setHead() ###

排他模式下，获得锁的节点，一定会被设置成头节点

private void setHead(Node node) {
        
        // 将head设置为当前ndoe
        head = node; 
        // 将获得锁的线程置null
        node.thread = null;
        node.prev = null;
    }

### shouldParkAfterFailedAcquire() ###

校验能否安全休眠：当前线程可以安心阻塞的标准，就是前一个节点线程状态是 SIGNAL 了

*  前一个结点是SIGNAL，return true
 *  前一个节点不是SIGNAL，return false（因为上层调用是自旋，所以最后一定能SIGNAL)
    
     *  waitStatus>0(已取消)：依次向前寻找，挂到一个未取消节点后面
     *  waitStatus<=0 但不是SIGNAL，就将前一个结点设为SIGNAL

// 入参 pred 是前一个节点，node 是当前节点。
    private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        
        int ws = pred.waitStatus;
        // 如果前一个节点 waitStatus 状态已经是 SIGNAL 了，直接返回，不需要在自旋了
        if (ws == Node.SIGNAL)
            return true;
        // 如果前一个节点状态已经被取消了
        if (ws > 0) {
        
            // 找到前一个状态不是取消的节点，因为把当前 node 挂在有效节点身上
            // 因为节点状态是取消的话是无效的，是不能作为 node 的前置节点的，所以必须找到 node 的有效节点才行
            do {
        
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        // 否则直接把前一个结点状态置 为SIGNAL
        } else {
        
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }

### parkAndCheckInterrupt() ###

通过park休眠当前线程，到时需要unpark唤醒

private final boolean parkAndCheckInterrupt() {
        
            LockSupport.park(this);
            return Thread.interrupted();
    }

![在这里插入图片描述][20200920171039589.png_pic_center]

这里最终调用的 Unsafe 类，是位于sun.misc包下的一个类，可以用来在任意内存地址位置处读写数据，支持一些CAS原子操作。Java 最初被设计为一种安全的受控环境。尽管如此，HotSpot 还是包含了一个后门 sun.misc.Unsafe，提供了一些可以直接操控内存和线程的底层操作。

park 及 unpark 底层其实是调用了**操作系统**的 Mutex互斥量、Condition 信号量、\_counter计数器

*  park主要流程：
    
     *  当\_counter > 0，则直接调用 pthread\_mutex\_unlock 解锁并返回；
     *  当超时时间 time>0，则调用 pthread\_cond\_timedwait 进行超时等待，直到超时时间到达；
     *  当超时时间 time=0，则调用 pthread\_cond\_wait 等待；
     *  当 wait 返回时设置\_counter = 0，并调用 pthread\_mutex\_unlock 解锁；
 *  unpark主要流程：
    
     *  调用pthread\_mutex\_lock获取锁，设置\_counter=1，调用pthread\_mutex\_unlock解锁；
     *  若\_counter的原值等于0，则调用pthread\_cond\_signal进行通知处理；

> PS：所以，并不是说 AQS 就全部都在用户态完成了，无论是 AQS 还是 synchronized 的阻塞，都需要借助操作系统底层的互斥量、信号量。**本质上都是线程的阻塞、唤醒，都会涉及线程状态切换**。关于 park 相关源码可以参考[这篇文章][Link 1]…

### cancelAcquire() ###

在node获得锁过程中失败或出现异常，就将 node 设为 CANCELLED，然后删除；并发下可能短暂维持

private void cancelAcquire(Node node) {
        
      	// 将无效节点过滤
    	if (node == null)
    		return;
      	// 设置该节点不关联任何线程，也就是虚节点
    	node.thread = null;
    	Node pred = node.prev;
      	// 通过前驱节点，跳过取消状态的node
    	while (pred.waitStatus > 0)
    		node.prev = pred = pred.prev;
      	// 获取过滤后的前驱节点的后继节点
    	Node predNext = pred.next;
      	// 把当前node的状态设置为CANCELLED
    	node.waitStatus = Node.CANCELLED;
      	// 如果当前节点是尾节点，将从后往前的第一个非取消状态的节点设置为尾节点
      	// 更新失败的话，则进入else，如果更新成功，将tail的后继节点设置为null
    	if (node == tail && compareAndSetTail(node, pred)) {
        
    		compareAndSetNext(pred, predNext, null);
    	} else {
        
    		int ws;
        	// 如果当前节点不是head的后继节点，1:判断当前节点前驱节点的是否为SIGNAL，
        	// 2:如果不是，则把前驱节点设置为SINGAL看是否成功
        	// 如果1和2中有一个为true，再判断当前节点的线程是否为null
        	// 如果上述条件都满足，把当前节点的前驱节点的后继指针指向当前节点的后继节点
    		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 {
        
          	// 如果当前节点是head的后继节点，或者上述条件不满足，那就唤醒当前节点的后继节点
    			unparkSuccessor(node);
    		}
    		node.next = node; // help GC
    	}
    }

## 2.独占-释放 ##

### release() ###

release() 是 unlock() 的基础方法，用于释放锁。如图是 Reentrantlock 的 unlock 方法，

![在这里插入图片描述][20200920171623251.png_pic_center]  
可以看到，是调用的继承了 AQS 的 Sync 的 release 方法，该方法是在 AQS 中实现的

public final boolean release(int arg) {
        
        // tryRelease 交给子类去实现
        // 一般就是用当前同步器状态减去 arg，如果返回 true 说明成功释放锁。
        if (tryRelease(arg)) {
         
            // 如果可以释放锁，保存头结点head
            Node h = head;
            // 头节点不为空，并且非初始化状态
            if (h != null && h.waitStatus != 0)
                // 唤醒同步队列队二
                unparkSuccessor(h);
            return true;
        }
        return false;
    }

下面是 ReentrantLock 的 Sync 的 tryRelease 方法，可以看到，核心是将 AQS 状态置为 0

protected final boolean tryRelease(int releases) {
        
        int c = getState() - releases; 
        if (Thread.currentThread() != getExclusiveOwnerThread())
            throw new IllegalMonitorStateException();
        boolean free = false;
        // 如果state=0了，就是可以释放锁了
        if (c == 0) {
         
            free = true; 
            // 将拿锁线程置为null
            setExclusiveOwnerThread(null); 
        }
        // 重置同步器的state
        setState(c); 
        // 返回是否成功释放
        return free; 
    }

### unparkSuccessor（核心方法) ###

找到真正的队二（不是CANCELLED状态)，并唤醒

*  head.next 非 null 非 CANCELLED，head.next 就是被唤醒对象
 *  head.next 如果不行，就从同步队列尾找，防止唤醒的 node 的前置节点仍然是 CANCELLED

> 注：此时并未删除 head

private void unparkSuccessor(Node node) {
        
        // node 节点是当前释放锁的节点，也是同步队列的头节点
        int ws = node.waitStatus;
        // 如果节点已经被取消了，把节点的状态置为初始化
        if (ws < 0)
            compareAndSetWaitStatus(node, ws, 0);
    
        // 拿出队二s
        Node s = node.next;
        // s 为空，表示 node 的后一个节点为空
        // s.waitStatus 大于0，代表 s 节点已经被取消了
        // 遇到以上这两种情况，就从队尾开始，向前遍历，找到第一个 waitStatus 字段不是被取消的
        if (s == null || s.waitStatus > 0) {
        
            s = null;
       		
            // 结束条件是前置节点就是head了
            for (Node t = tail; t != null && t != node; t = t.prev)
                // t.waitStatus <= 0 说明 t 当前没有被取消，肯定还在等待被唤醒
                if (t.waitStatus <= 0)
                    s = t;
        }
        // 唤醒以上代码找到的线程
        if (s != null)
            LockSupport.unpark(s.thread);
    }

到这里需要明确一点，在AQS中，**若一个线程释放了锁，接下来只会唤醒一个线程，并不是把所有线程都唤醒**，然后大家再去竞争。

[JUC_JUC_AQS]: https://blog.csdn.net/weixin_43935927/article/details/108693296
[JUC_JUC_AQS 1]: https://blog.csdn.net/weixin_43935927/article/details/113901484
[JUC_JUC_AQS 2]: https://blog.csdn.net/weixin_43935927/article/details/113904384
[JUC_JUC_AQS 3]: https://blog.csdn.net/weixin_43935927/article/details/113901502
[JUC_JUC_AQS 4]: https://blog.csdn.net/weixin_43935927/article/details/108699927
[20200920155545794.png_pic_center]: /images/20221024/8de3866ffbfd48d2ad62e509a92c9efd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl80MzkzNTkyNw_size_16_color_FFFFFF_t_70_pic_center]: /images/20221024/4a9be29c1ee34ebab777f4a85570f097.png
[20200920163515366.png_pic_center]: /images/20221024/4f65bc652ab04e938d425c5ecb3327a7.png
[20200920171039589.png_pic_center]: /images/20221024/2da039b080b24683af9ae5c0439a602d.png
[Link 1]: https://www.jianshu.com/p/f32f941335aa
[20200920171623251.png_pic_center]: /images/20221024/3eed5b46947641d9a16ce4cd472e4df3.png