AQS源码分析总结

AQS是并发编程的一个最基本组件,是一个抽象同步器。
网上有很多详细介绍AQS的博文,在这里我就不仔细介绍了,主要写一些重要的内容。
AQS中重要的几个属性:

//同步队列的头节点
private transient volatile Node head;
//同步队列的尾节点
private transient volatile Node tail;
//同步状态
private volatile int state;

由于一个共享资源同一时间可以被一条线程持有,也可以被多个线程持有,因此AQS中存在两种模式,共享模式独占模式

  • 共享模式是共享状态值state每次可以由多个线程持有,如CountDownLatchSemaphore
  • 独占模式是共享状态值state每次只能由一条线程持有,其他线程如果需要获取,则需要阻塞。如ReentrantLock

    独占锁的获取

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

先尝试获取锁,获取失败会调用addWaiter将当前线程添加到同步队列,之后队列中的每个节点会调用acquireQueued()方法通过自旋的方式先再尝试获取一下锁,如果失败,将当前节点的前驱节点的状态设置为SIGNAL,并将该线程阻塞,并判断该线程是否被中断。如果被中断了,当前节点获取锁后进行中断操作。
这里用到了模版方法的设计模式,tryAcquire是一个抽象方法,具体实现需要到子类中去完成。

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

下面详细介绍获取独占锁失败后,添加到队列的过程,调用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;
            if (compareAndSetTail(pred, node)) {
                pred.next = node;
                return node;
            }
        }
        enq(node);
        return node;
    }

先创建出一个节点
1) 当前尾节点不为空,采用尾插法利用CAS机制将新创建的节点添加到尾部,设置为尾节点。如果添加失败,就执行enq()方法。
2) 如果尾节点为空,调用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;
                }
            }
        }
    }

enq是一个自旋操作,
1) 如果尾节点为空,说明当前线程是第一个加入同步队列的,先用CAS操作新添一个头节点head,并将尾节点指向它。第二次循环会跳往另一个执行区域。
2) 利用CAS操作将该节点添加到尾部。
直到自旋添加成功,就结束循环。

入队成功后,就要为该节点开启自旋,尝试获得锁。

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);
        }
    }

先获取当前节点的先驱节点
1) 如果先驱接节点是头节点并且成功获取同步状态的时候,将当前节点设置为头节点,然后将之前的头节点的next指针设置为null并且pre指针也为null,即将前节点与队列断开,
2)如果获取失败,就调用shouldParkAfterFailedAcquire方法,主要作用是将该节点的前驱节点的状态设置为SIGNAL,表示线程阻塞。

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        int ws = pred.waitStatus;
        if (ws == Node.SIGNAL)
            return true;
        if (ws > 0) {
            do {
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else {
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }

shouldParkAfterFailedAcquire方法中,如果前驱节点的状态是SIGNAL,就直接返回true;如果状态值大于0,表明该前驱节点已经被取消,则将该前驱节点去除掉,向前移;其他情况,将该前驱节点设置为SIGNAL。如果添加失败,就返回false,因为是自旋,下一次再尝试。
由于acquireQueued方法是一个循环,在第二次执行到shouldParkAfterFailedAcquire方法时,由于0号节点的waitStatus已经为Node.SIGNAL了,所以shouldParkAfterFailedAcquire方法会返回true,然后继续执行parkAndCheckInterrupt方法,将该线程已经阻塞,并怕判断该线程是否中断。

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

独占锁的释放

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

独占锁通过tryRelease释放成功后,如果头节点head不为null,并且状态值不为0,就会对它的后继节点进行唤醒。

private void unparkSuccessor(Node node) {
    int ws = node.waitStatus;
    if (ws < 0)
        compareAndSetWaitStatus(node, ws, 0);
    //头节点的后继节点
    Node s = node.next;
    if (s == null || s.waitStatus > 0) {
        s = null;
        for (Node t = tail; t != null && t != node; t = t.prev)
            if (t.waitStatus <= 0)
                s = t;
    }
    if (s != null)
        //后继节点不为null时唤醒该线程
        LockSupport.unpark(s.thread);
}

如果头节点的后继节点是空或者它的状态值大于0,表明它是失效的,就要从尾节点节点向前查找,找到最后一个状态值小于等于0的节点,然后对该节点进行唤醒。

共享锁的获取与释放

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

共享锁可以同时被多个线程拥有,可以在初始设置的时候将state设置为大于0的值,每一个线程获取一次,就减1,当state大于等于0时,别的线程也能够拥有该锁,当小于0时,就不可以,在共享锁模式下,当前线程拿到锁后,会直接通知后继节点去拿锁,而不必等待锁被释放的时候再通知。 在锁释放的时候,支持多个线程释放同步线程同步状态。
参考文章:
深入理解AbstractQueuedSynchronizer(AQS)
java并发编程系列:牛逼的AQS(上)

上一篇:《动手学习深度学习》之三:3.RNN循环神经网络(进阶)-4种模型(打卡2.6)


下一篇:3 softmax