AQS源码解读

Posted on | Edited on | In , | Views:

AbstractQueueSynchronizer

定义了一套多线程访问共享资源的同步框架
首先维护了一个volatile state状态量 和 一个阻塞队列,是FIFO队列,队列使用链表实现.state的访问方式有三种

  • getstate()
  • setState()
  • compareAndSetState()
    AQS 定义了两种资源共享状态,一种是独占(Exclusive 如ReentrantLock),一种是共享(Share可以同时多个线程执行).
    不同的自定义同步实现器在实现时只需要实现共享资源state的获取和释放.至于具体的等待队列的维护与修改,顶层已了.自定义经实现好同步器主要实现一下几个方法.
  • isHeldExclusive():表示该线程是否独占资源,只有用到condition才需要实现
  • tryAcquire():独占方式,尝试去获取资源,成功返回true,失败放回false
  • tryRelease():独占方式:尝试去释放资源
  • tryAcquireShared(int): 共享方式,尝试获取资源,负数表示失败,0表示成功,但没有剩余资源,正数表示成功并且有资源
  • tryReleaseShared(int): 共享方式:尝试释放资源,如果释放后允许释放后续等待资源返回true,否则放回false

以ReentrantLock为例,首先初始化state为0,表示未锁定状态.一旦调用lock.lock(),会尝试调用tryAcquire()并且将state+1,此时其他线程会获取锁失败,知道unlock()讲state置为0,在获取到此锁的时候,可以继续获取,但是每次都会是state+1.
CountDownLatch 中,任务分为N个子线程执行,state初始化为N,这个N个子线程是并行执行的,每个子线程执行完之后会countDown()一次,state会通过CAS减一,等待所有的线程都执行完之后,在unpark()会调用主线程继续执行自己的工作.

Node 节点

队列是通过Node来保持一个队列的,就相当于链表中的普通节点,保存了同步的线程以及线程的状态,上一个节点,下一个节点.:

  • int waitStatus 共有四种状态:
    • CANCELLED: 值为1,在等待过程中被中断,需要从队列中取消该节点.
    • SIGNAL :值为-1,后继节点的线程处于等待状态,当前节点的线程如果释放了同步状态或取消,将会通知后集节点,使后继节点的线程得意运行.
    • CONDITION :值为-2,表示该节点处于等待队列中，节点的线程等待在condition上,当其他线程调用Condition.signal()方法会唤醒此线程,进入到同步队列中
    • PROPAGATE：值为-3，与共享模式相关，在共享模式中，该状态标识结点的线程处于可运行状态。
  • Node prev,前驱节点,当节点加入同步队列时被设置
  • Node next,后继节点,
  • Node nextWaiter 等待队列中的后继节点.如果当前节点是共享的,那么这个字段是一个SHARED常量,也就是说节点类型(独占和共享)和等待队列中的后继节点共用同一个字段
  • Thread thread 获取同步状态的线程.

    方法一览

    isShared() 判断是否是共享模式
    1
    2
    3
    final boolean isShared() {
        return nextWaiter == SHARED;
    }

predecessor() 返回前驱节点,如果前驱为空,抛出异常

1
2
3
4
5
6
7
final Node predecessor() throws NullPointerException {
        Node p = prev;
        if (p == null)
            throw new NullPointerException();
        else
            return p;
   }

AQS内部属性

  • Node head: 队列的头结点
  • Node tail: 队列的尾节点
  • int state: 锁的状态,独占模式下为0或1,0为空闲,1为已经有线程获取到锁.共享模式下为还剩余多少资源
  • long spinForTimeoutThreshold = 1000L: 最小的等待时间 在下面讲到

方法 独占模式方法

acquire() 获取锁

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

TryAcquire(arg)如果成功获取锁,则直接返回.而tryAcquire方法需要具体实现
addWaiter() 方法则将线程加入到队列的尾部,并标记为独占模式
acquireQueued() 使线程在等待队列中获取资源,知道获取资源后返回,如果被中断过,则返回true.

addWaiter()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
 1 private Node addWaiter(Node mode) {
 2     //以给定模式构造结点。mode有两种：EXCLUSIVE（独占）和SHARED（共享）
 3     Node node = new Node(Thread.currentThread(), mode);
 4     
 5     //尝试快速方式直接放到队尾。
 6     Node pred = tail;
 7     if (pred != null) {
 8         node.prev = pred;
 9         if (compareAndSetTail(pred, node)) {//利用CAS进行替换
10             pred.next = node;
11             return node;
12         }
13     }
14     
15     //上一步失败则通过enq入队。
16     enq(node);
17     return node;
18 }

enq(Node)

如果直接添加到队尾失败则使用enq方法加入队尾

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
 1 private Node enq(final Node node) {
 2     //CAS"自旋"，直到成功加入队尾
 3     for (;;) {
 4         Node t = tail;
 5         if (t == null) { // 队列为，空创建一个空的标志结点作为head结点，并将tail也指向它。
 6             if (compareAndSetHead(new Node()))
 7                 tail = head;
 8         } else {//正常流程，放入队尾
 9             node.prev = t;
10             if (compareAndSetTail(t, node)) {
11                 t.next = node;
12                 return t;
13             }
14         }
15     }
16 }

使用CAS自旋,直到成功加入队尾

acquireQueued(Node int)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
 1 final boolean acquireQueued(final Node node, int arg) {
 2     boolean failed = true;//标记是否成功拿到资源
 3     try {
 4         boolean interrupted = false;//标记等待过程中是否被中断过
 5         
 6         //又是一个“自旋”！
 7         for (;;) {
 8             final Node p = node.predecessor();//拿到前驱
 9             //如果前驱是head，即该结点已成老二，那么便有资格去尝试获取资源（可能是老大释放完资源唤醒自己的，当然也可能被interrupt了）。
10             if (p == head && tryAcquire(arg)) {
11                 setHead(node);//拿到资源后，将head指向该结点。所以head所指的标杆结点，就是当前获取到资源的那个结点或null。
12                 p.next = null; // setHead中node.prev已置为null，此处再将head.next置为null，就是为了方便GC回收以前的head结点。也就意味着之前拿完资源的结点出队了！
13                 failed = false;
14                 return interrupted;//返回等待过程中是否被中断过
15             }
16             
17             //如果自己可以休息了，就进入waiting状态，直到被unpark()
18             if (shouldParkAfterFailedAcquire(p, node) &&
19                 parkAndCheckInterrupt()) //parkAndCheckiInterrupt方法会进行检查是否被中断过,并且阻塞线程,此处用自旋是为了保证唤醒的节点能够同步的获取到锁.
20                 interrupted = true;//如果等待过程中被中断过，哪怕只有那么一次，就将interrupted标记为true
21         }
22     } finally {
23         if (failed)
24             cancelAcquire(node);
25     }
26 }

shouldParkAfterFailedAcquire

用于检查状态,看看自己是否可以去休息,也就是阻塞

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
 1 private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
 2     int ws = pred.waitStatus;//拿到前驱的状态
 3     if (ws == Node.SIGNAL)
 4         //如果已经告诉前驱拿完号后通知自己一下，那就可以安心休息了
 5         return true;
 6     if (ws > 0) {
 7         /*
 8          * 如果前驱放弃了，那就一直往前找，直到找到最近一个正常等待的状态，并排在它的后边。
 9          * 注意：那些放弃的结点，由于被自己“加塞”到它们前边，它们相当于形成一个无引用链，稍后就会被保安大叔赶走了(GC回收)！
10          */
11         do {
12             node.prev = pred = pred.prev;
13         } while (pred.waitStatus > 0);
14         pred.next = node;
15     } else {
16          //如果前驱正常，那就把前驱的状态设置成SIGNAL，告诉它拿完号后通知自己一下。有可能失败，人家说不定刚刚释放完呢！
17         compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
18     }
19     return false;
20 }

###　parkAndCheckInterrupt()　
将进程阻塞，并且返回中断状态.

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

doAcquireInterruptibly(int arg)

此方法为支持中断的获取资源,如果在过程中发生了中断,则方法会抛出中断并且返回

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
private void doAcquireInterruptibly(int arg)
        throws InterruptedException {
        final Node node = addWaiter(Node.EXCLUSIVE);
        boolean failed = true;
        try {
            for (;;) {
                final Node p = node.predecessor();
                if (p == head && tryAcquire(arg)) {//首先看是否能够获取资源,获取不到则不会取消
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return;
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    throw new InterruptedException();
            }
        } finally {
            if (failed)//如果failed=true,说明没有获取到资源,并且有过中断,因为中断后会返回并且抛出异常.则取消该节点.
                cancelAcquire(node);
        }
    }

doAcquireNanos(int arg, long nanosTimeout)

是支持超时的获取同步状态,在时间限制到的情况下会直接返回false,否则获取到同步状态则返回true

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
private boolean doAcquireNanos(int arg, long nanosTimeout)
            throws InterruptedException {
        if (nanosTimeout <= 0L) 
            return false;
        final long deadline = System.nanoTime() + nanosTimeout;//计算出deadline时间
        final Node node = addWaiter(Node.EXCLUSIVE);//将此节点加入队列
        boolean failed = true;
        try {
            //自旋
            for (;;) {
                final Node p = node.predecessor();//获取前驱,与普通获取一样
                if (p == head && tryAcquire(arg)) {
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return true;
                }
                nanosTimeout = deadline - System.nanoTime(); //再次判断是否超时
                if (nanosTimeout <= 0L)
                    return false;
                if (shouldParkAfterFailedAcquire(p, node) &&//判断是否可以阻塞
                    nanosTimeout > spinForTimeoutThreshold)//必须得经过一次spinForTimeoutThreshold
                    LockSupport.parkNanos(this, nanosTimeout);
                    //如果nanosTimeout小于spinForTimeoutThreshold,将不会是线程进入超时等待,则会进入一个自旋的过程.原因在于非常短的超时无法做到十分精确,如果这个时候在进行超时等待,会让nanoTimeout的超时从整体上变现的不精确.
                if (Thread.interrupted())
                    throw new InterruptedException();//如果线程被中断过,则会抛出一个异常并且放回
            }
        } finally {
            if (failed)//如果没有获得资源或则被中断或者超时都会取消此节点.
                cancelAcquire(node);
        }
    }

realease
会释放指定量的资源,首先执行tryrelease()方法,此方法为继承者实现.如果彻底释放,会唤醒队列中的其他线程来获取资源

1
2
3
4
5
6
7
8
9
1 public final boolean release(int arg) {
2     if (tryRelease(arg)) {
3         Node h = head;    //找到头结点
4         if (h != null && h.waitStatus != 0)//在头结点不为空的情况下并且waitStatus不为0也就是不是刚初始化的情况
5             unparkSuccessor(h);    //唤醒等待队列里的下一个线程
6         return true;
7     }
8     return false;
9 }

unparkSuccessor()

用来唤醒等待对垒中的下一个线程

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
 1 private void unparkSuccessor(Node node) {
 2     //这里，node一般为当前线程所在的结点。
 3     int ws = node.waitStatus;
 4     if (ws < 0)//置零当前线程所在的结点状态，允许失败。
 5         compareAndSetWaitStatus(node, ws, 0);
 6 
 7     Node s = node.next;//找到下一个需要唤醒的结点s
 8     if (s == null || s.waitStatus > 0) {//如果为空或已取消,则从尾节点开始找,知道找到第一个有效的节点
 9         s = null;
10         for (Node t = tail; t != null && t != node; t = t.prev)
11             if (t.waitStatus <= 0)//从这里可以看出，<=0的结点，都是还有效的结点。
12                 s = t;
13     }
14     if (s != null)
15         LockSupport.unpark(s.thread);//唤醒
16 }

共享模式

和独占模式差不多,但是在共享模式只有当前线程是头结点的下一个节点的时候,才可以去获取资源,有剩余的话会唤醒后续的节点,不会隔着来.

acquireShared(int arg)

1
2
3
4
public final void acquireShared(int arg) {
        if (tryAcquireShared(arg) < 0)//尝试直接获取,小于0说明没有资源
            doAcquireShared(arg);
    }

doAcquireShared(int arg)

与独占模式相同,只不过在加入队列的节点模式改变成了共享模式,还有一点就是在共享模式下如果还有剩余资源则会继续唤醒之后的节点.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
private void doAcquireShared(int arg) {
    // 首先以共享方式加入队列
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        boolean interrupted = false;//标志位检查是否呗中断过
        // 与独占模式相同,自旋保证获取资源,
        for (;;) {
            final Node p = node.predecessor();
            if (p == head) {
                int r = tryAcquireShared(arg);
                if (r >= 0) {
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    if (interrupted)
                        selfInterrupt();
                    failed = false;
                    return;
                }
            }
            //加入阻塞队列之后进行中断检查,如果中断过则取消节点.
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                interrupted = true;
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

doAcquireSharedInterruptibly(int arg)

这里与独占模式不同,独占模式首先获取资源,将头结点设置为当前节点,获取不到则阻塞,并且被中断过就抛出异常,在这里首先会获取资源,设置头结点之后多了一条propagate操作.在这个会在判断后继节点是否为shared的,如果是,则会唤醒.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
private void doAcquireSharedInterruptibly(int arg)
    throws InterruptedException {
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        for (;;) {
            final Node p = node.predecessor();
            if (p == head) {
                int r = tryAcquireShared(arg);
                if (r >= 0) {//如果还有资源,进入下一个方法
                    setHeadAndPropagate(node, r);
                    p.next = null; // help GC
                    failed = false;
                    return;
                }
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                throw new InterruptedException();
        }
    } finally {
        if (failed)
            cancelAcquire(node);
    }
}

setHeadAndPropagate(node, r)

在这里会先看后继节点,如果后继节点是shared状态则会继续唤醒后集节点

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
    private void setHeadAndPropagate(Node node, int propagate) {
        Node h = head; // Record old head for check below
        setHead(node);
        if (propagate > 0 || h == null || h.waitStatus < 0 ||
            (h = head) == null || h.waitStatus < 0) {
            Node s = node.next;
            if (s == null || s.isShared())
                doReleaseShared();
        }
    }
~~~ 

### doReleaseShared()
首先先看头结点,如果头结点为signal状态,则会唤醒头结点的下一个节点.如果状态为0,则讲状态转换为propagate状态,继续向下传播.
~~~java
private void doReleaseShared() {
        
        for (;;) {
            Node h = head;
            if (h != null && h != tail) {
                int ws = h.waitStatus;
                if (ws == Node.SIGNAL) {
                    if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
                        continue;            // loop to recheck cases
                    unparkSuccessor(h);//在这里会唤醒h节点之后的一个等待节点
                }
                else if (ws == 0 &&
                         !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
                    continue;                // loop on failed CAS
            }
            if (h == head)                   // loop if head changed
                break;
        }
    }

unparkSuccessor(Node node)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
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)
        LockSupport.unpark(s.thread);
}

cancelAcquire(Node node)

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
    private void cancelAcquire(Node node) {
    if (node == null) //节点为空则直接返回
        return;

    node.thread = null; //将节点的线程置位空,方便GC
    Node pred = node.prev;// 找到前驱节点
    while (pred.waitStatus > 0)//这里找到第一个有效的前驱节点
        node.prev = pred = pred.prev;
    Node predNext = pred.next;
    node.waitStatus = Node.CANCELLED; // 将此节点状态设置为取消状态

    //如果节点为尾节点,并且设置尾节点为最后一个有效节点,则把尾节点的next设置为null//有助于GC
    if (node == tail && compareAndSetTail(node, pred)) {
        compareAndSetNext(pred, predNext, null);
    } else {//否则将前驱节点替换为node的后继节点
        int ws;
        if (pred != head &&
            ((ws = pred.waitStatus) == Node.SIGNAL ||
             (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&
            pred.thread != null) {//如果前驱节点不为为头结点并且前驱节点的状态为signal而且线程不为空.
            Node next = node.next;
            if (next != null && next.waitStatus <= 0)
                compareAndSetNext(pred, predNext, next);//将前驱有效节点的下一个节点设置为node的下一个节点
        } else {
            unparkSuccessor(node);//如果前驱节点是头结点,就唤醒node之前的前驱节点.
        }
        node.next = node; // help GC
    }
}

内部类ConditionObject

实现 Condition接口和serializable接口
condition的实现主要包括等待队列,等待和通知.等待队列是一个FIFO的队列,如果线程调用了Condition.await()方法,则会释放锁,构造成节点加入等待队列并进入等待状态.这里节点复用了同步器的节点定义.

属性:

  • Node firstWaiter: 第一个等待节点
  • Node lastWaiter: 最后一个等待节点

主要方法:

await()

调用await()方法回事当前线程进入等待队列并且释放锁,同事状态变为等待状态.在返回时,当前线程一定获取了Condition相关联的锁.
从队列的角度看await()方法,调用await()方法,相当于同步队列的首节点移动到了Condition的等待队列中

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
//因为此时节点还获取有锁,所以不会产生竞争,直接赋值就可以
public final void await() throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    //当前线程加入等待队列中
    Node node = addConditionWaiter();
    //释放同步状态
    int savedState = fullyRelease(node);
    int interruptMode = 0;
    while (!isOnSyncQueue(node)) { //这里判断是否应该等待,看节点状态是否为Condition状态
        LockSupport.park(this);//阻塞
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
    }
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)//加入竞争中
        interruptMode = REINTERRUPT;
    if (node.nextWaiter != null) // clean up if cancelled
        unlinkCancelledWaiters();
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);
}

###　addConditionWaiter()
将节点加入到condition等带队列中.

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
private Node addConditionWaiter() {
    Node t = lastWaiter;
    // If lastWaiter is cancelled, clean out.
    if (t != null && t.waitStatus != Node.CONDITION) {
        unlinkCancelledWaiters();//断开已经取消的节点链
        t = lastWaiter;
    }
    Node node = new Node(Thread.currentThread(), Node.CONDITION);//创建一个新的节点.
    if (t == null)
        firstWaiter = node;
    else
        t.nextWaiter = node;
    lastWaiter = node;
    return node; 
}

unlinkCancelledWaiters()

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
private void unlinkCancelledWaiters() {
    Node t = firstWaiter;
    Node trail = null;
    while (t != null) {
        Node next = t.nextWaiter;
        if (t.waitStatus != Node.CONDITION) {
            t.nextWaiter = null;
            if (trail == null)
                firstWaiter = next;
            else
                trail.nextWaiter = next;
            if (next == null)
                lastWaiter = trail;
        }
        else
            trail = t;
        t = next;
    }
}

signal()

唤醒第一个节点,如果当前线程没有获取锁,则抛出异常

1
2
3
4
5
6
7
public final void signal() {
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    Node first = firstWaiter;
    if (first != null)
        doSignal(first);
}

doSignal()

1
2
3
4
5
6
7
8
private void doSignal(Node first) {//循环唤醒下一个线程
    do {
        if ( (firstWaiter = first.nextWaiter) == null)
            lastWaiter = null;
        first.nextWaiter = null;
    } while (!transferForSignal(first) &&
             (first = firstWaiter) != null);
}

transferForSignal(Node first)

将节点设置状态为0,然后在加入队尾,在唤醒线程

1
2
3
4
5
6
7
8
9
final boolean transferForSignal(Node node) {
    if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
        return false;
    Node p = enq(node);
    int ws = p.waitStatus;
    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
        LockSupport.unpark(node.thread);//这里唤醒线程后,线程会从await()方法的while循环中退出(isOnSyncQueue(Node node)方法返回ture,节点已经在同步队列中)借着在调用同步器的acquireQueued()方法加入竞争
    return true;
}

其他方法:

  • hasQueuedThreads() 返回是否有阻塞线程
  • hasContended 返回是否有线程在执行
  • isQueued(Thread thread) 返回当前线程是否在阻塞中
  • getQueueLength() 返回队列的长度
  • Collection getQueuedThreads 返回正在阻塞的线程
  • getExclusiveQueuedThreads() 返回独占状态的阻塞线程,返回一个线程集合Collection
  • getSharedQueuedThreads() 返回共享状态下的线程集合