AQS

2021.02.14 07:02 422
阅读约 3 分钟

什么是AQS

全称是 AbstractQueuedSynchronizer,是阻塞式锁和相关的同步器工具的框架,AQS使用一个先入先出的队列表示排队等待锁的线程,队列头节点称作哨兵节点或者哑节点,它不与任何线程关联。其他的节点与等待线程关联。

AQS中还有一个表示状态的字段state,例如ReentrantLock用它来表示线程重入锁的次数,Semphore用它表示剩余的许可数量,FutureTask用它表示任务的状态。对state变量值的更新都采用CAS操作保证更新操作的原子性。

特点:

  • 用state属性来表示资源状态(分独占模式和共享模式),子类需要定义如何维护状态,控制如何获取和释放锁。
    • getState- 获取state状态
    • setState- 设置state状态
    • CAS- 乐观锁机制设置state状态
    • 独占模式是只有一个线程能够访问资源,而共享模式可以允许多个线程访问
  • 提供了基于FIFO的等待队列,类似于Monitor的 EntryList (组设等待队列)
  • 条件变量来实现等待,唤醒机制,支持多个条件变量,类似于Monitor的WaitSet。

子类主要实现这样的一些方法

  • tryAcquire
  • tryRelease
  • tryAcquireShared
  • tryReleaseShared
  • isHeldExclusively

目标

AQS要实现的功能目标

  • 阻塞版本获取锁acquire 和非阻塞版本尝试获取锁tryAcquire
  • 获取锁超时机制
  • 通过打断取消机制
  • 独占机制及共享机制
  • 条件变量

设计

AQS的基本思想很简单

获取锁的逻辑

while(state 状态不允许获取) {
 if(队列中还没有此线程) {
 入队并阻塞
 }
}
当前线程出队

释放锁的逻辑

if(state 状态允许了) {
 恢复阻塞的线程(s)
}

要点

  • 原子维护state变量
  • 阻塞及恢复线程
  • 维护队列

state设计

  • state 使用volatile 配合 cas 保证其修改时的原子性
  • state 使用了 32bit int来维护同步状态

阻塞恢复

  • 使用 park & unpark 来实现线程的暂停和恢复,具体原理在之前讲过了,先 unpark 再 park 也没 问题
  • park & unpark 是针对线程的,而不是针对同步器的,因此控制粒度更为精细
  • park 线程还可以通过 interrupt 打断

队列设计

  • 使用了 FIFO 先入先出队列,并不支持优先级队列
  • 设计时借鉴了 CLH 队列,它是一种单向无锁队列

队列中有 head 和 tail 两个指针节点,都用 volatile 修饰配合 cas 使用,每个节点有 state 维护节点

入队伪代码,只需要考虑 tail 赋值的原子性 

do {
 // 原来的 tail
 Node prev = tail;
 // 用 cas 在原来 tail 的基础上改为 node
} while(tail.compareAndSet(prev, node))

出队伪代码

// prev 是上一个节点
while((Node prev=node.prev).state != 唤醒状态) {
}
// 设置头节点
head = node;

NonfairSync实现

我们从源代码出发,分析非公平锁获取锁和释放锁的过程。 

lock()

lock()源码分析

final void lock() {
    if (compareAndSetState(0, 1))
        setExclusiveOwnerThread(Thread.currentThread());
    else
        acquire(1);
}

首先使用CAS操作,判断state是否是0(表示当前锁未被占用),如果是0则把它置为1,并且设置当前线程为该锁的独占线程,表示获取锁成功。当多个线程同时尝试占用同一个锁时,CAS操作只能保证一个线程操作成功,剩下的只能乖乖的去排队啦。

若当前有三个线程去竞争锁,假设线程A的cas操作成功了,拿到了锁,那么线程B和C则设置state失败,走到了else里面。我们往下看acquire。

acquire(arg)

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

1.tryAcquire(arg) 尝试获取锁

如果尝试获取锁成功,方法直接返回。

final boolean nonfairTryAcquire(int acquires) {
    //获取当前线程
    final Thread current = Thread.currentThread();
    //获取state变量值
    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");
        // 更新state值为新的重入次数
        setState(nextc);
        return true;
    }
    //获取锁失败
    return false;
}

非公平锁tryAcquire(arg)流程:检查state字段,若为0,表示未被占用,可以尝试占用。若已被占用,将占用线程与当前线程比较,若被当前线程占用,则更新state字段,表示重入锁的次数。如果以上两点都没有成功,则获取锁失败,返回false。

2.第二步,入队

线程A已经占用了锁,所以B和C执行tryAcquire(arg) 失败,并且入队等待,如果线程A拿着锁死死不放,那么B和C就会被挂起。

先看获取锁失败时执行的addWaiter(Node.EXCLUSIVE)

/**
 * 将新节点和当前线程关联并且入队列
 * @param mode 独占/共享
 * @return 新节点
 */
private Node addWaiter(Node mode) {
    //初始化节点,设置关联线程和模式(独占 or 共享)
    Node node = new Node(Thread.currentThread(), mode);
    // 获取尾节点引用
    Node pred = tail;
    // 尾节点不为空,说明队列已经初始化过
    if (pred != null) {
        node.prev = pred;
        // 设置新节点为尾节点
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
    // 尾节点为空,说明队列还未初始化,需要初始化head节点并入队新节点
    enq(node);
    return node;
}

B、C线程同时尝试入队列,由于队列尚未初始化,tail==null,故至少会有一个线程会走到enq(node)。我们假设同时走到了enq(node)里。

/**
 * 初始化队列并且入队新节点
 */
private Node enq(final Node node) {
    //开始自旋
    for (;;) {
        Node t = tail;
        if (t == null) { // Must initialize
            // 如果tail为空,则新建一个head节点,并且tail指向head
            if (compareAndSetHead(new Node()))
                tail = head;
        } else {
        	 // 设置新节点的前驱节点为tail节点
            node.prev = t;
            // CAS设置tail节点为新节点
            if (compareAndSetTail(t, node)) {
            // 将原来的tail节点的next节点设置为新的tail节点
                t.next = node;
                return t;
            }
        }
    }
}

这里体现了cas自旋来实现原子操作。由于compareAndSetHead的实现使用了unsafe类提供的CAS操作,所以只有一个线程会创建head节点成功。假设线程B成功,之后B、C开始第二轮循环,此时tail已经不为空,两个线程都走到else里面。假设B线程compareAndSetTail成功,那么B就可以返回了,C由于入队失败还需要第三轮循环。最终所有线程都可以成功入队。

第三步,挂起

B和C相继执行acquireQueued(final Node node, int arg)。这个方法让已经入队的线程尝试获取锁,若失败则会被挂起。

/**
 * 已经入队的线程尝试获取锁
 node 为返回的tail节点
 */
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节点出队,在某个时间点被GC回收
                failed = false; //获取成功
                return interrupted; //返回是否被中断过
            }
            // 判断获取失败后是否可以挂起,若可以则挂起
            if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                // 线程若被中断,设置interrupted为true
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

假设B和C在竞争锁的过程中A一直持有锁,那么它们的tryAcquire操作都会失败,因此会走到第2个if语句中。我们再看下shouldParkAfterFailedAcquire和parkAndCheckInterrupt都做了哪些事吧。

/**
 * 判断当前线程获取锁失败之后是否需要挂起.
 */
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    //前驱节点的状态
    int ws = pred.waitStatus;
    if (ws == Node.SIGNAL)
        // 前驱节点状态为signal -1,返回true直接挂起
        return true;
    // 前驱节点状态为CANCELLED
    if (ws > 0) {
        // 从队尾向前寻找第一个状态不为CANCELLED的节点
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        // 将前驱节点的状态设置为SIGNAL-1
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}
  
/**
 * 挂起当前线程,返回线程中断状态并重置
 */
private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this);
    return Thread.interrupted();
}

当前线程进入 acquireQueued 逻辑 

1. acquireQueued 会在一个死循环中不断尝试获得锁,失败后进入 park 阻塞 

2. 如果自己是紧邻着 head(排第二位),那么再次 tryAcquire 尝试获取锁,当然这时 state 仍为 1,失败 

3. 进入 shouldParkAfterFailedAcquire 逻辑,将前驱 node,即 head 的 waitStatus 改为 -1,这次返回 false 

shouldParkAfterFailedAcquire会先判断当前节点的前驱是否状态符合要求,若符合则返回true,然后调用parkAndCheckInterrupt,将自己挂起。如果不符合,再看前驱节点是否>0(CANCELLED),若是那么向前遍历直到找到第一个符合要求的前驱,若不是则将前驱节点的状态设置为SIGNAL。

  整个流程中,如果前驱结点的状态不是SIGNAL,那么自己就不能安心挂起,需要去找个安心的挂起点,同时可以再尝试下看有没有机会去尝试竞争锁。

unlock()

 unlock相对于lock就简单很多。源码如下:

public void unlock() {
    sync.release(1);
}
  
public final boolean release(int arg) {
    if (tryRelease(arg)) {
        Node h = head;
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h);
        return true;
    }
    return false;
}

先尝试释放锁,若释放成功,那么查看头结点的状态是否为SIGNAL,如果是则唤醒头结点的下个节点关联的线程,如果释放失败那么返回false表示解锁失败。这里我们也发现了,每次都只唤起头结点的下一个节点关联的线程。

我们再看下tryRelease的执行过程:

/**
 * 释放当前线程占用的锁
 * @param releases
 * @return 是否释放成功
 */
protected final boolean tryRelease(int releases) {
    // 计算释放后state值
    int c = getState() - releases;
    // 如果不是当前线程占用锁,那么抛出异常
    if (Thread.currentThread() != getExclusiveOwnerThread())
        throw new IllegalMonitorStateException();
    boolean free = false;
    if (c == 0) {
        // 锁被重入次数为0,表示释放成功
        free = true;
        // 清空独占线程
        setExclusiveOwnerThread(null);
    }
    // 更新state值
    setState(c);
    return free;
}

这里入参为1。tryRelease的过程为:当前释放锁的线程若不持有锁,则抛出异常。若持有锁,计算释放后的state值是否为0,若为0表示锁已经被成功释放,并且则清空独占线程,最后更新state值,返回free。 

用一张流程图总结一下非公平锁的获取锁的过程。 

尝试实现不可重入锁 

// 自定义锁(不可重入锁)
class MyLock implements Lock {

    // 独占锁  同步器类
    class MySync extends AbstractQueuedSynchronizer {
        @Override
        protected boolean tryAcquire(int arg) {
            System.out.println(arg);
            if(compareAndSetState(0, 1)) {
                // 加上了锁,并设置 owner 为当前线程
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            return false;
        }

        @Override
        protected boolean tryRelease(int arg) {
            System.out.println(arg);
            setExclusiveOwnerThread(null);
            setState(0);
            return true;
        }

        @Override // 是否持有独占锁
        protected boolean isHeldExclusively() {
            return getState() == 1;
        }

        public Condition newCondition() {
            return new ConditionObject();
        }
    }

    private MySync sync = new MySync();

    @Override // 加锁(不成功会进入等待队列)
    public void lock() {
        sync.acquire(1);
    }

    @Override // 加锁,可打断
    public void lockInterruptibly() throws InterruptedException {
        sync.acquireInterruptibly(1);
    }

    @Override // 尝试加锁(一次)
    public boolean tryLock() {
        return sync.tryAcquire(1);
    }

    @Override // 尝试加锁,带超时
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        return sync.tryAcquireNanos(1, unit.toNanos(time));
    }

    @Override // 解锁
    public void unlock() {
        sync.release(1);
    }

    @Override // 创建条件变量
    public Condition newCondition() {
        return sync.newCondition();
    }
}

 

字数:652 发布于 7 个月前
Copyright 2018-2021 Siques