深入浅出理解Java并发AQS的共享锁模式

叁歲伎倆 2023-09-23 19:29 64阅读 0赞

### 自定义共享锁例子 ###

首先我们通过AQS实现一个非常最最最轻量简单的共享锁例子，帮助大家对共享锁有一个整体的感知。

@Slf4j
    public class ShareLock {
    
        /**
         * 共享锁帮助类
         */
        private static class ShareSync extends AbstractQueuedSynchronizer {
    
            private int lockCount;
    
            /**
             * 创建共享锁帮助类，最多有count把共享锁，超过了则阻塞
             *
             * @param count 共享锁数量
             */
            public ShareSync(int count) {
               this.lockCount = count;
            }
    
            /**
             * 尝试获取共享锁
             *
             * @param arg 每次获取锁的数量
             * @return 返回正数，表示后续其他线程获取共享锁可能成功； 返回0，表示后续其他线程无法获取共享锁；返回负数，表示当前线程获取共享锁失败
             */
            @Override
            protected int tryAcquireShared(int arg) {
                // 自旋
                for (;;) {
                    int c = getState();
                    // 如果持有锁的数量大于指定数量，返回-1，线程进入阻塞
                    if(c >= lockCount) {
                        return -1;
                    }
                    int nextc = c + 1;
                    // cas设置成功，返回1，获取到共享锁
                    if (compareAndSetState(c, nextc)) {
                        return 1;
                    }
                }
            }
    
            /**
             * 尝试释放共享锁
             *
             * @param arg 释放锁的数量
             * @return 如果释放后允许唤醒后续等待结点返回true，否则返回false
             */
            @Override
            protected boolean tryReleaseShared(int arg) {
                // 自旋操作
                for (; ; ) {
                    int c = getState();
                    // 如果没有锁了
                    if (c == 0) {
                        return false;
                    }
                    // 否则锁量-1
                    int nextc = c - 1;
                    // cas修改状态
                    if (compareAndSetState(c, nextc)) {
                        return true;
                    }
                }
            }
        }
    
        private final ShareSync sync;
    
        public ShareLock(int count) {
            this.sync = new ShareSync(count);
        }
    
        /**
         * 加共享锁
         */
        public void lockShare() {
            sync.acquireShared(1);
        }
    
        /**
         * 释放共享锁
         */
        public void releaseShare() {
            sync.releaseShared(1);
        }
    }
    复制代码

*  创建内部类共享帮助锁`ShareSync`类，继承自`AbstractQueuedSynchronizer`类，实现了共享锁相关的方法`tryAcquireShared()`和`tryReleaseShared()`。
 *  创建`ShareLock`，提供了`lockShare()`加锁和`releaseShare()`两个API。

**验证：**

public static void main(String[] args) throws InterruptedException {
            ShareLock shareLock = new ShareLock(3);
            for (int i = 0; i < 5; i++) {
                new Thread(() -> {
                    shareLock.lockShare();
                    try {
                        log.info("lock success");
                        Thread.sleep(1000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    } finally {
                        shareLock.releaseShare();
                        log.info("release success");
                    }
    
                }, "thread-" + i).start();
            }
            Thread.sleep(10000);
        }
    复制代码

*  一共创建最多共同有3个线程共享的共享锁。
 *  创建5个线程去竞争共享锁。

**运行结果：**

![format_png][]

*  运行结果显示每次最多只有3个`lock success`，说明同时只有3个线程共享。
 *  只有在释放共享锁以后，其他线程才能获取锁。

下面对它的实现原理一探究竟。

### 核心原理机制 ###

共享模式也是由AQS提供的，首先我们关注下AQS的数据结构。

![format_png 1][]

AQS内部维护了一个volatile int state（代表共享资源）和一个FIFO线程等待队列（多线程争用资源被阻塞时会进入此队列）。

AQS作为一个抽象方法，提供了加锁、和释放锁的框架，这里采用的模板方模式，在上面中提到的`tryAcquireShared`、`tryReleaseShared`就是和共享模式相关的模板方法。

<table> 
 <thead> 
  <tr> 
   <th><strong>方法名</strong></th> 
   <th><strong>描述</strong></th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>protected int tryAcquireShared(int arg)</td> 
   <td>共享方式。arg为获取锁的次数，尝试获取资源。负数表示失败；0表示成功，但没有剩余可用资源；正数表示成功，且有剩余资源。</td> 
  </tr> 
  <tr> 
   <td>protected boolean tryReleaseShared(int arg)</td> 
   <td>共享方式。arg为释放锁的次数，尝试释放资源，如果释放后允许唤醒后续等待结点返回True，否则返回False。</td> 
  </tr> 
 </tbody> 
</table>

共享模式的入口方法如下：

<table> 
 <thead> 
  <tr> 
   <th><strong>方法名</strong></th> 
   <th><strong>描述</strong></th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>void acquireShared(int arg)</td> 
   <td>共享模式获取锁，不响应中断。</td> 
  </tr> 
  <tr> 
   <td>void acquireSharedInterruptibly(int arg)</td> 
   <td>共享模式获取锁，响应中断。</td> 
  </tr> 
  <tr> 
   <td>tryAcquireSharedNanos(int arg, long nanosTimeout)</td> 
   <td>尝试在共享模式下获取锁，如果中断则中止，如果超过给定超时则失败。</td> 
  </tr> 
  <tr> 
   <td>boolean releaseShared(int arg)</td> 
   <td>共享模式下释放锁。</td> 
  </tr> 
 </tbody> 
</table>

### 源码解析 ###

![format_png 2][]

上图是AQS的类结构图，其中标红部分是组成AQS的重要成员变量。

#### 成员变量 ####

1.  **state共享变量**

AQS中里一个很重要的字段state，表示同步状态，是由volatile修饰的，用于展示当前临界资源的获锁情况。通过getState()，setState()，compareAndSetState()三个方法进行维护。

关于state的几个要点：

*  使用volatile修饰，保证多线程间的可见性。
 *  getState()、setState()、compareAndSetState()使用final修饰，限制子类不能对其重写。
 *  compareAndSetState()采用乐观锁思想的CAS算法，保证原子性操作。

1.  **CLH队列（FIFO队列）**

AQS里另一个重要的概念就是CLH队列，它是一个双向链表队列，其内部由head和tail分别记录头结点和尾结点，队列的元素类型是Node。

private transient volatile Node head;
    private transient volatile Node tail;
    复制代码

Node的结构如下：

static final class Node {
        //共享模式下的等待标记
        static final Node SHARED = new Node();
        //独占模式下的等待标记
        static final Node EXCLUSIVE = null;
        //表示当前结点已取消调度。当timeout或被中断（响应中断的情况下），会触发变更为此状态，进入该状态后的结点将不会再变化。
        static final int CANCELLED =  1;
        //表示后继结点在等待当前结点唤醒。后继结点入队时，会将前继结点的状态更新为SIGNAL。
        static final int SIGNAL    = -1;
        //表示结点等待在Condition上，当其他线程调用了Condition的signal()方法后，CONDITION状态的结点将从等待队列转移到同步队列中，等待获取同步锁。
        static final int CONDITION = -2;
        //共享模式下，前继结点不仅会唤醒其后继结点，同时也可能会唤醒后继的后继结点。
        static final int PROPAGATE = -3;
        //状态，包括上面的四种状态值，初始值为0，一般是节点的初始状态
        volatile int waitStatus;
        //上一个节点的引用
        volatile Node prev;
        //下一个节点的引用
        volatile Node next;
        //保存在当前节点的线程引用
        volatile Thread thread;
        //condition队列的后续节点
        Node nextWaiter;
    }
    复制代码

注意，waitSstatus负值表示结点处于有效等待状态，而正值表示结点已被取消。所以源码中很多地方用>0、<0来判断结点的状态是否正常。

1.  **exclusiveOwnerThread**

AQS通过继承AbstractOwnableSynchronizer类，拥有的属性。表示独占模式下同步器持有的线程。

#### 共享锁获取acquireShared(int) ####

acquireShared(int)是共享锁模式下线程获取共享资源的入口方法，它会获取指定量的资源，获取成功则直接返回，获取失败则进入等待队列，直到获取到资源为止，整个过程无法响应中断。

public final void acquireShared(int arg) {
        if (tryAcquireShared(arg) < 0)
            doAcquireShared(arg);
    }
    复制代码

方法的整体流程如下：

1.  tryAcquireShared()尝试获取资源，需要自定义同步器去实现，返回负值代表获取失败；0代表获取成功，但没有剩余资源；正数表示获取成功，还有剩余资源，其他线程还可以去获取。
2.  如果失败则通过doAcquireShared()进入等待队列，直到获取到资源为止才返回。

**doAcquireShared(int)**

此方法用于将当前线程加入等待队列尾部休息，直到其他线程释放资源唤醒自己，自己成功拿到相应量的资源后才返回。

private void doAcquireShared(int arg) {
        //封装线程为共享Node 加入队列尾部
        final Node node = addWaiter(Node.SHARED);
        //是否成功标志
        boolean failed = true;
        try {
            //等待过程中是否被中断过的标志
            boolean interrupted = false;
            // 自旋操作
            for (;;) {
                // 获取前驱节点
                final Node p = node.predecessor();
                //如果到head的下一个，因为head是拿到资源的线程，此时node被唤醒，很可能是head用完资源来唤醒自己的
                if (p == head) {
                    //尝试获取资源
                    int r = tryAcquireShared(arg);
                    //成功
                    if (r >= 0) {
                        //将head指向自己，还有剩余资源可以再唤醒之后的线程
                        setHeadAndPropagate(node, r);
                        p.next = null; // help GC
                        //如果等待过程中被打断过，此时将中断补上。
                        if (interrupted)
                            selfInterrupt();
                        failed = false;
                        return;
                    }
                }
    
                //判断状态，寻找安全点，进入waiting状态，等着被unpark()或interrupt()
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }
    复制代码

`doAcquireShared`方法的实现和获取独占锁中的`acquireQueued`方法很类似，但是主要有一点不同，那就是线程在被唤醒后，若成功获取到了共享锁，还需要判断共享锁是否还能被其他线程获取，若可以，则继续向后唤醒它的下一个节点对应的线程。

**setHeadAndPropagate(Node, int)**

该方法主要将当前节点设置为头节点，同时判断条件是否符合（比如还有剩余资源），还会去唤醒后继结点，毕竟是共享模式。

private void setHeadAndPropagate(Node node, int propagate) {
        Node h = head;
        //head指向自己
        setHead(node);
         //如果还有剩余量，继续唤醒下一个邻居线程
        if (propagate > 0 || h == null || h.waitStatus < 0) {
            Node s = node.next;
            if (s == null || s.isShared())
                // 唤醒操作
                doReleaseShared();
        }
    }
    复制代码

#### 共享释放releaseShared(int) ####

`releaseShared(int)`是共享模式下线程释放共享资源的入口，它会释放指定量的资源，如果成功释放且允许唤醒等待线程，它会唤醒等待队列里的其他线程来获取资源。

public final boolean releaseShared(int arg) {
        //尝试释放资源
        if (tryReleaseShared(arg)) {
            //唤醒后继结点
            doReleaseShared();
            return true;
        }
        return false;
    }
    复制代码

方法的整体流程如下：

*  tryReleaseShared尝试释放锁，这由自定义同步器去实现， 返回true表示释放成功。
 *  doReleaseShared唤醒后续队列中等待的节点，

**doReleaseShared()**

此方法主要用于唤醒队列中等待的共享节点。

private void doReleaseShared() {
        // 自旋操作
        for (;;) {
            // 获取头节点
            Node h = head;
            if (h != null && h != tail) {
                // 获取节点的等待状态
                int ws = h.waitStatus;
                // 如果节点等待状态是-1， -1表示有责任唤醒后续节点的状态
                if (ws == Node.SIGNAL) {
                    // cas修改当前节点的等待状态为0
                    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                        continue;
                    //唤醒后续节点
                    unparkSuccessor(h);
                }
                else if (ws == 0 &&
                         !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue;
            }
            if (h == head)// head发生变化
                break;
        }
    }
    复制代码

*  逻辑是一个死循环，每次循环中重新读取一次head，然后保存在局部变量h中，再配合`if(h == head) break;`，这样，循环检测到head没有变化时就会退出循环。注意，head变化一定是因为：acquire thread被唤醒，之后它成功获取锁，然后setHead设置了新head。而且注意，只有通过`if(h == head) break;`即head不变才能退出循环，不然会执行多次循环。
 *  `if (h != null && h != tail)`判断队列是否至少有两个node，如果队列从来没有初始化过（head为null），或者head就是tail，那么中间逻辑直接不走，直接判断head是否变化了。
 *  如果队列中有两个或以上个node，那么检查局部变量h的状态：

*   *  如果状态为SIGNAL，说明h的后继是需要被通知的。通过对CAS操作结果取反，将`compareAndSetWaitStatus(h, Node.SIGNAL, 0)`和`unparkSuccessor(h)`绑定在了一起。说明了只要head成功得从SIGNAL修改为0，那么head的后继的代表线程肯定会被唤醒了。
     *  如果状态为0，说明h的后继所代表的线程已经被唤醒或即将被唤醒，并且这个中间状态即将消失，要么由于acquire thread获取锁失败再次设置head为SIGNAL并再次阻塞，要么由于acquire thread获取锁成功而将自己（head后继）设置为新head并且只要head后继不是队尾，那么新head肯定为SIGNAL。所以设置这种中间状态的head的status为PROPAGATE，让其status又变成负数，这样可能被被唤醒线程检测到。

*  如果状态为PROPAGATE，直接判断head是否变化。
 *  两个continue保证了进入那两个分支后，只有当CAS操作成功后，才可能去执行if(h == head) break;，才可能退出循环。
 *  if(h == head) break;保证了，只要在某个循环的过程中有线程刚获取了锁且设置了新head，就会再次循环。目的当然是为了再次执行unparkSuccessor(h)，即唤醒队列中第一个等待的线程。

### 总结 ###

本文主要讲解了AQS的共享模式，通过一个自定义简单的demo帮助大家深入浅出的理解，同时深入分析了源码实现，希望对大家有帮助。

[format_png]: https://img-blog.csdnimg.cn/img_convert/50f7df6f1b8259ce0408c99ca001eed9.webp?x-oss-process=image/format,png
[format_png 1]: https://img-blog.csdnimg.cn/img_convert/f3e009cea9d0749310edd819f0307e94.webp?x-oss-process=image/format,png
[format_png 2]: https://img-blog.csdnimg.cn/img_convert/cfcc399ae7e92da0d1f9d1b0958f9feb.webp?x-oss-process=image/format,png