当前位置： 首页 > news >正文

Java面试--AQS

news 来源：原创 2024/9/22 1:39:07

一、AQS概述

AQS 全名 AbstractQueuedSynchronizer，意为抽象队列同步器，JUC（java.util.concurrent 包）下面的 Lock 和其他一些并发工具类都是基于它来实现的。AQS 维护了一个 volatile 的 state 和一个 CLH（FIFO）双向队列

在这里插入图片描述

二、分析

2.1、state

AQS 使用一个 int 成员变量来表示同步状态，通过内置的 FIFO 队列来完成获取资源线程的排队工作。AQS 使用 CAS 对该同步状态进行原子操作实现对其值的修改

private volatile int state;   // 共享变量，使用 volatile 修饰保证线程可见性

状态信息通过 protected 类型的 getState()，setState()，compareAndSetState() 进行操作

// 返回同步状态的当前值
protected final int getState() {
	return state;
}

// 设置同步状态的值
protected final void setState(int newState) {
    state = newState;
}

// 原子地（CAS操作）将同步状态值设置为给定值 update 如果当前同步状态的值等于 expect（期望值）
protected final boolean compareAndSetState(int expect, int update) {
    return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}

2.2、AQS对资源的共享方式

1、Exclusive（独占）

只有一个线程能执行，如 ReentrantLock。又可分为公平锁和非公平锁，ReentrantLock 同时支持两种锁，下面以 ReentrantLock 对这两种锁的定义做介绍：

  • 公平锁 ：按照线程在队列中的排队顺序，先到者先拿到锁
  • 非公平锁 ：当线程要获取锁时，先通过两次 CAS 操作去抢锁，如果没抢到，当前线程再加入到队列中等待唤醒。

2、Share（共享）

多个线程可同时执行，如 Semaphore/CountDownLatch。Semaphore、CountDownLatch、 CyclicBarrier、ReadWriteLock 我们都会在后面讲到。

ReentrantReadWriteLock 可以看成是组合式，因为 ReentrantReadWriteLock 也就是读写锁允许多个线程同时对某一资源进行读。

不同的自定义同步器争用共享资源的方式也不同。自定义同步器在实现时只需要实现共享资源 state 的获取与释放方式即可，至于具体线程等待队列的维护（如获取资源失败入队/唤醒出队等），AQS 已经在上层已经帮我们实现好了

2.3、AQS 底层使用了模板方法模式

同步器的设计是基于模板方法模式的，如果需要自定义同步器一般的方式是这样（模板方法模式很经典的一个应用）：

  1. 使用者继承 AbstractQueuedSynchronizer 并重写指定的方法。（这些重写方法很简单，无非是对于共享资源 state 的获取和释放）
  2. 将 AQS 组合在自定义同步组件的实现中，并调用其模板方法，而这些模板方法会调用使用者重写的方法。

这和我们以往通过实现接口的方式有很大区别，这是模板方法模式很经典的一个运用

AQS 使用了模板方法模式，自定义同步器时需要重写下面几个 AQS 提供的钩子方法：

// 独占方式。尝试获取资源，成功则返回 true，失败则返回 false
protected boolean tryAcquire(int)

// 独占方式。尝试释放资源，成功则返回 true，失败则返回 false
protected boolean tryRelease(int)

// 共享方式。尝试获取资源。负数表示失败；0 表示成功，但没有剩余可用资源；正数表示成功，且有剩余资源
protected int tryAcquireShared(int)

// 共享方式。尝试释放资源，成功则返回 true，失败则返回 false
protected boolean tryReleaseShared(int)

// 该线程是否正在独占资源。只有用到 condition 才需要去实现它
protected boolean isHeldExclusively()

什么是钩子方法呢？ 钩子方法是一种被声明在抽象类中的方法，一般使用 protected 关键字修饰，它可以是空方法（由子类实现），也可以是默认实现的方法。模板设计模式通过钩子方法控制固定步骤的实现

除了上面提到的钩子方法之外，AQS 类中的其他方法都是 final ，所以无法被其他类重写

以 ReentrantLock 为例

state 初始化为 0，表示未锁定状态。A 线程 lock() 时，会调用 tryAcquire() 独占该锁并将 state + 1 。此后，其他线程再 tryAcquire() 时就会失败，直到 A 线程 unlock() 到 state=0（即释放锁）为止，其它线程才有机会获取该锁。当然，释放锁之前，A 线程自己是可以重复获取此锁的（state 会累加），这就是可重入的概念。但要注意，获取多少次就要释放多少次，这样才能保证 state 是能回到零态的

再以 CountDownLatch 以例

任务分为 N 个子线程去执行，state 也初始化为 N（注意 N 要与线程个数一致）。这 N 个子线程是并行执行的，每个子线程执行完后 countDown() 一次，state 会 CAS(Compare and Swap) 减 1。等到所有子线程都执行完后(即 state=0 )，会 unpark() 主调用线程，然后主调用线程就会从 await() 函数返回，继续后余动作

一般来说，自定义同步器要么是独占方法，要么是共享方式，他们也只需实现 tryAcquire-tryRelease、tryAcquireShared-tryReleaseShared中的一种即可。但 AQS 也支持自定义同步器同时实现独占和共享两种方式，如ReentrantReadWriteLock

2.4、CLH（FIFO）队列

AQS 中是通过内部类 Node 来维护一个 CLH 队列的。源码如下：

static final class Node {
    /** 标记共享式访问 */
    static final Node SHARED = new Node();
    
    /** 标记独占式访问 */
    static final Node EXCLUSIVE = null;

    /** 字段 waitStatus 的值，表示当前节点已取消等待 */
    static final int CANCELLED =  1;
    
    /**字段 waitStatus 的值，表示当前节点取消或释放资源后，通知下一个节点 */
    static final int SIGNAL    = -1;
    
    /** 表示正在等待触发条件 */
    static final int CONDITION = -2;
    /**
     * 表示下一个共享获取应无条件传播
     */
    static final int PROPAGATE = -3;

    /**
     *   SIGNAL:     表示当前节点取消或释放资源后，通知下一个节点
     *   CANCELLED:  表示当前节点已取消等待
     *   CONDITION:  表示正在等待触发条件
     *   PROPAGATE:  表示下一个共享获取应无条件传播
     */
    volatile int waitStatus;

    /**
     * 前节点
     */
    volatile Node prev;

    /**
     * 下一个节点
     */
    volatile Node next;

    /**
     * 节点对应线程
     */
    volatile Thread thread;

    /**
     * 下一个等待的节点
     */
    Node nextWaiter;

    /**
     * 是否是共享式访问
     */
    final boolean isShared() {
        return nextWaiter == SHARED;
    }

    /**
     * 返回前节点
     */
    final Node predecessor() throws NullPointerException {
        Node p = prev;
        if (p == null)
            throw new NullPointerException();
        else
            return p;
    }

    Node() {    // 共享式访问的构造函数
    }

    Node(Thread thread, Node mode) {     // 用于被添加等待者使用
        this.nextWaiter = mode;
        this.thread = thread;
    }

    Node(Thread thread, int waitStatus) { // 用于Condition使用
        this.waitStatus = waitStatus;
        this.thread = thread;
    }
}

下面来看 ReentrantLock 中相关的源代码：

ReentrantLock 默认采用非公平锁，因为考虑获得更好的性能，通过 boolean 来决定是否用公平锁（传入 true 用公平锁）

private final Sync sync;

public ReentrantLock() {
    // 默认非公平锁
    sync = new NonfairSync();
}

public ReentrantLock(boolean fair) {
    sync = fair ? new FairSync() : new NonfairSync();
}

写段代码往下走

ReentrantLock reentrantLock = new ReentrantLock();
reentrantLock.lock();

reentrantLock 调用的是

public void lock() {
   sync.lock();
}

ReentrantLock 中公平锁的 lock 方法

static final class FairSync extends Sync {
    final void lock() {
        acquire(1);
    }
}

调用的是 AQS 中 acquire 方法

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

1、如果获取到资源使用权，则返回 true，反之 fasle。如果获取到资源，返回 true，!true 为 false，根据 && 的短路性，则不会执行后续方法，直接跳过程序。如果未获取到资源，返回 false，!false 为 true，则进入后续方法

往下走，看下的 tryAcquire 方法

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

这个方法需要 ReentantLock 自己实现

protected final boolean tryAcquire(int acquires) {
   final Thread current = Thread.currentThread();
    int c = getState();
    if (c == 0) {
        // 1. 和非公平锁相比，这里多了一个判断：是否有线程在等待
        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;
}

非公平锁的 lock 方法

static final class NonfairSync extends Sync {
    final void lock() {
        // 2. 和公平锁相比，这里会直接先进行一次 CAS，成功就返回了
        if (compareAndSetState(0, 1))
            setExclusiveOwnerThread(Thread.currentThread());
        else
            acquire(1);
    }
}

acquire 方法一样，都是 AQS 中的

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

往下走，看下子类实现的 tryAcquire 方法

protected final boolean tryAcquire(int acquires) {
	return nonfairTryAcquire(acquires);
}

继续，nonfairTryAcquire 方法

final boolean nonfairTryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();
    if (c == 0) {
        // 这里没有对阻塞队列进行判断
        if (compareAndSetState(0, acquires)) {
            setExclusiveOwnerThread(current);
            return true;
        }
    }
    else if (current == getExclusiveOwnerThread()) {
        int nextc = c + acquires;
        if (nextc < 0) // overflow
            throw new Error("Maximum lock count exceeded");
        setState(nextc);
        return true;
    }
    return false;
}

总结：公平锁和非公平锁只有两处不同：

  • 非公平锁在调用 lock 后，首先就会调用 CAS 进行一次抢锁，如果这个时候恰巧锁没有被占用，那么直接就获取到锁返回了
  • 非公平锁在 CAS 失败后，和公平锁一样都会进入到 tryAcquire 方法，在 tryAcquire 方法中，如果发现锁这个时候被释放了（state == 0），非公平锁会直接 CAS 抢锁，但是公平锁会判断等待队列是否有线程处于等待状态，如果有则不去抢锁，乖乖排到后面

公平锁和非公平锁就这两点区别，如果这两次 CAS 都不成功，那么后面非公平锁和公平锁是一样的，都要进入到阻塞队列等待唤醒

相对来说，非公平锁会有更好的性能，因为它的吞吐量比较大。当然，非公平锁让获取锁的时间变得更加不确定，可能会导致在阻塞队列中的线程长期处于饥饿状态

2、获取锁失败后，会执行 addWaiter(Node.EXCLUSIVE) 加入等待队列，具体实现方法如下

private Node addWaiter(Node mode) {
   // 封装当前线程和独占模式
   Node node = new Node(Thread.currentThread(), mode);
   
   // 获取尾部节点
   Node pred = tail;
   
   if (pred != null) {
       node.prev = pred;
       // CAS 设置尾部节点
       if (compareAndSetTail(pred, node)) {
           // 将尾节点的下一节点指向当前 node
           pred.next = node;
           return node;
       }
   }
   // 如果尾结点为空或者设置尾结点失败
   enq(node);
   return node;
}

compareAndSetTail 方法

private final boolean compareAndSetTail(Node expect, Node update) {
	return unsafe.compareAndSwapObject(this, tailOffset, expect, update);
}

主要的流程如下：

  • 通过当前的线程和锁模式新建一个节点
  • Pred 指针指向尾节点 Tail
  • 将 New 中 Node 的 Prev 指针指向 Pred
  • 通过 compareAndSetTail 方法，完成尾节点的设置。这个方法主要是对 tailOffset 和 Expect 进行比较，如果 tailOffset 的 Node 和 Expect 的 Node 地址是相同的，那么设置 Tail 的值为 Update 的值
// java.util.concurrent.locks.AbstractQueuedSynchronizer

static {
	try {
		stateOffset = unsafe.objectFieldOffset(AbstractQueuedSynchronizer.class.getDeclaredField("state"));
		headOffset = unsafe.objectFieldOffset(AbstractQueuedSynchronizer.class.getDeclaredField("head"));
		tailOffset = unsafe.objectFieldOffset(AbstractQueuedSynchronizer.class.getDeclaredField("tail"));
		waitStatusOffset = unsafe.objectFieldOffset(Node.class.getDeclaredField("waitStatus"));
		nextOffset = unsafe.objectFieldOffset(Node.class.getDeclaredField("next"));
	} catch (Exception ex) { 
    throw new Error(ex); 
  }
}

从 AQS 的静态代码块可以看出，都是获取一个对象的属性相对于该对象在内存当中的偏移量，这样我们就可以根据这个偏移量在对象内存当中找到这个属性。tailOffset 指的是 tail 对应的偏移量，所以这个时候会将 new 出来的 Node 置为当前队列的尾节点。同时，由于是双向链表，也需要将前一个节点指向尾节点

  • 如果 Pred 指针是 Null（说明等待队列中没有元素），或者当前 Pred 指针和 Tail 指向的位置不同（说明被别的线程已经修改），就需要看一下 Enq 的方法
private Node enq(final Node node) {
   for (;;) {
       Node t = tail;
       if (t == null) { // Must initialize
           if (compareAndSetHead(new Node()))
               tail = head;
       } else {
           node.prev = t;
           if (compareAndSetTail(t, node)) {
               t.next = node;
               return t;
           }
       }
   }
}

相关文章：

  • 电脑重装系统win11如何更改默认下载路径
  • MTI运动传感器ROS配置
  • DFS——剪枝优化迭代加深
  • 【Flutter】packages思维以及使用Java添加Android平台特定的实现在Flutter框架里的体现和运用
  • 无向图以及图的java代码实现
  • 大数据平台之Hadoop复习详细知识点
  • 四、网络层（五）IP组播
  • 513. 找树左下角的值
  • python csv模块读取/写入csv文件
  • Nacos学习笔记 （5）Nacos整合SpringBoot流程
  • 在服务器安装jupyter并在本地访问
  • MySQL索引-索引的优势和劣势
  • 基于R语言、MaxEnt模型融合技术的物种分布模拟、参数优化方法、结果分析制图与论文写作
  • 关于居住办公人口的统计技术解决方案
  • 四、网络层（三）IPv4
  • Docker+Jenkins+Gitee+Maven构建后台jar包后配置SSH传送到服务器并执行指定命令
  • 职场经验：游戏测试的主要工作及主要流程
  • 创建react项目
  • 信息安全产品认证
  • 2021地理设计组一等奖：面向游客的旅游路线优化设计——以丹霞山景区为例