AQS原理剖析_aqs waitstatus

AQS结构剖析

• 双向链表 + waitStatus的int值

锁的结构：

1. 实现Lock接口
2. 组合AQS进行并发状态控制

为什么使用双向链表实现？

因为链表移除和添加比较方便，只需要改动prev和next节点的指向即可，移除和添加都只需要操作一次，时间复杂度为O(1)。如果使用数组去实现，随着数据量的增加每次操作需要移动的次数也会更重

waitStatus的int值是什么？有什么用？

waitStatus

volatile int waitStatus AQS核心实现，等待状态，它有几种状态值：CANCELLED、SIGNAL、CONDITION、PROPAGATE

static final class Node {
        /** Marker to indicate a node is waiting in shared mode */
        static final Node SHARED = new Node();
        /** Marker to indicate a node is waiting in exclusive mode */
        static final Node EXCLUSIVE = null;

        /** waitStatus value to indicate thread has cancelled */
        static final int CANCELLED =  1;
        /** waitStatus value to indicate successor's thread needs unparking */
        static final int SIGNAL    = -1;
        /** waitStatus value to indicate thread is waiting on condition */
        static final int CONDITION = -2;
		/**
         * waitStatus value to indicate the next acquireShared should
         * unconditionally propagate
         */
        static final int PROPAGATE = -3;

    	// AQS核心实现，等待状态
        volatile int waitStatus;

        volatile Node prev;

        volatile Node next;

        volatile Thread thread;

        Node nextWaiter;

       
    }
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

**CANCELLED：**由锁状态变成取消状态，这个时候就可以被gc回收了

SIGNAL: 插入名为4的节点到3和2之间，然后将4节点的前继节点也就是2的waitStatus改成SIGNAL状态

其余的节点相信大家直接看释义就能明白了

源码分析

参考流程图，我们按照程序流程来分析源代码

ReentrantLock

ReentrantLock支持公平锁和非公平锁，我们可以通过它的构造函数来控制选择哪种锁。默认无参构造是非公平锁实现

ReentrantLock提供的公平锁FairSync和非公平锁NonfairSync都是继承自AbstractQueuedSynchronizer就是AQS

public class ReentrantLock implements Lock, java.io.Serializable {
    private static final long serialVersionUID = 7373984872572414699L;
    /** Synchronizer providing all implementation mechanics */
    private final Sync sync;
    
    /**
     * 默认无参构造非公平锁实现
     * Creates an instance of {@code ReentrantLock}.
     * This is equivalent to using {@code ReentrantLock(false)}.
     */
    public ReentrantLock() {
        sync = new NonfairSync();
    }

    /**
     * 通过boolean fair控制选择公平锁和非公平锁
     * Creates an instance of {@code ReentrantLock} with the
     * given fairness policy.
     *
     * @param fair {@code true} if this lock should use a fair ordering policy
     */
    public ReentrantLock(boolean fair) {
        sync = fair ? new FairSync() : new NonfairSync();
    }

    /**
     * Base of synchronization control for this lock. Subclassed
     * into fair and nonfair versions below. Uses AQS state to
     * represent the number of holds on the lock.
     */
    abstract static class Sync extends AbstractQueuedSynchronizer {
        private static final long serialVersionUID = -5179523762034025860L;
        
        /**
         * lock加锁方法，从这里作为入口开始分析
         * Performs {@link Lock#lock}. The main reason for subclassing
         * is to allow fast path for nonfair version.
         */
        abstract void lock();
        
        ......
        
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
35
36
37
38
39
40
41
42

lock

abstract void lock();是模板方法模式，由子类实现，在ReentrantLock中它的实现有公平锁和非公平锁两者，这里我们只有关注非公平锁的实现

先来比较一下公平锁和非公锁的区别在哪？

重点看lock()方法

非公平锁NonfairSync.lock()它一上来就先通过if (compareAndSetState(0, 1))cas去抢锁

​ 如果抢锁成功：

​ 则把当前线程，也就是自身，通过setExclusiveOwnerThread设置为当前独占锁的线程（占用锁的线程）

​ 如果抢锁失败：

​ 则走acquire(1)方法，继续抢锁，在失败就通过enq加入阻塞队列队尾

公平锁FairSync.lock()，则是直接调用acquire(1)方法，内部实现大致是先通过状态判断有无线程正在占用，如果没有也就是state == 0则继续通过hasQueuedPredecessor判断当前线程前面有没有其它等待的线程，如果没有在去抢锁，如果有则返回false，通过enq加入阻塞队列队尾

独占锁：同一时刻只有一个线程可以持有锁，其它线程未获取到锁时，会被阻塞

/**
     * Sync object for non-fair locks
     */
    static final class NonfairSync extends Sync {
        private static final long serialVersionUID = 7316153563782823691L;

        /**
         * Performs lock.  Try immediate barge, backing up to normal
         * acquire on failure.
         */
        final void lock() {
            if (compareAndSetState(0, 1))
                setExclusiveOwnerThread(Thread.currentThread());
            else
                acquire(1);
        }

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

 	/**
     * Sync object for fair locks
     */
    static final class FairSync extends Sync {
        private static final long serialVersionUID = -3000897897090466540L;

        final void lock() {
            acquire(1);
        }

        /**
         * Fair version of tryAcquire.  Don't grant access unless
         * recursive call or no waiters or is first.
         */
        protected final boolean tryAcquire(int acquires) {
            final Thread current = Thread.currentThread();
            int c = getState();
            if (c == 0) {
                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;
        }
    }

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
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

tryAcquire

我们继续分析非公平锁的实现

程序调用了acquire(1)之后，会先通过tryAcquire(1)去尝试获取锁，重点看一下它的实现

它是先通过int c = getState()获取锁标记，判断是否有锁

​ 如果锁状态等于0，那说明无锁

​ 则去通过cas抢锁，抢锁成功，则把自己设置为独占锁的线程

​ 如果锁状态不等于0，说明有锁

​ 先走else if的判断当前线程和独占锁的线程是否为同一线程，如果是，则直接拿到锁，也就是重入锁的特性，ReentrantLock就是重入独占锁，拿到锁之后继续给state累加1，表示有锁

​ 如果else if也判断失败，则返回false，tryAcquire尝试获取锁失败，这时走acquireQueued(addWaiter(Node.EXCLUSIVE), arg)

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

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

/**
* Performs non-fair tryLock.  tryAcquire is implemented in
* subclasses, but both need nonfair try for trylock method.
*/
final boolean nonfairTryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    // State: 锁标记 0是无锁、大于等于1是有锁状态()
    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;
}

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

加入队列队尾

在addWaiter中，能看到它是先创建了一个当前线程的node节点，然后获取到了tail节点，也就是尾节点，如果tail节点存在，那么则将当前线程创建的node节点的prev，也就是当前线程的前置节点指向现有的tail尾节点

然后通过cas抢锁，抢锁成功

​ 把自己设置为尾节点，在把之前的尾节点的next指向现在的node节点，并返回node节点出去

抢锁失败

​ 则通过enq方法，自旋加入队列。简单的说enq之前的代码是一种快速尝试插入节点，加入队列队尾的方法

​ 那么为什么需要enq自旋入队列呢？

​ 因为在这里是存在锁竞争的，所以需要抢锁，在操作

	/**
	 *	当前线程入队列，并返回当前线程对应的node节点
     * Creates and enqueues node for current thread and given mode.
     *
     * @param mode Node.EXCLUSIVE for exclusive, Node.SHARED for shared
     * @return the new node
     */
    private Node addWaiter(Node mode) {
        // 以给定模式构造节点。mode有两种：EXCLUSVIE（独占）和SHARED（共享）
        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
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

/**
     * Inserts node into queue, initializing if necessary. See picture above.
     * @param node the node to insert
     * @return node's predecessor
     */
    private Node enq(final Node node) {
        // CAS"自旋"，直到成功加入队尾
        for (;;) {
            Node t = tail;
            if (t == null) { // Must initialize
                // 队列为空，创建一个空的标志节点作为head节点，并将tail也指向它
                // 创建第一个节点，头尾都是自己
                if (compareAndSetHead(new Node()))
                    tail = head;
            } else { // 正常流程，加入队尾
                node.prev = t;
                if (compareAndSetTail(t, node)) {
                    t.next = node;
                    return t;
                }
            }
        }
    }

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

判断前驱节点释放为head

/**
     * Acquires in exclusive uninterruptible mode for thread already in
     * queue. Used by condition wait methods as well as acquire.
     *
     * @param node the node
     * @param arg the acquire argument
     * @return {@code true} if interrupted while waiting
     */
    final boolean acquireQueued(final Node node, int arg) {
        boolean failed = true; // 标记是否成功拿到资源
        try {
            boolean interrupted = false; // 标记等待过程中是否被中断过
            // 自旋
            for (;;) {
           		// 当前节点的前驱节点
                final Node p = node.predecessor();
                // 如果前驱节点是head，尝试获取资源（可能是head释放完资源唤醒当前线程），当然也可能被interrupt)
                if (p == head && tryAcquire(arg)) {
                    // 竞争锁成功
                    // 设置当前线程为head节点
                    setHead(node);
                    // 出队
                    p.next = null; // help GC
                    failed = false; // 成功获取资源
                    return interrupted; // 返回等待过程中是否被中断过
                }
                // park，挂起线程
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(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
30
31
32
33
34
35
36

接下来我们看shouldParkAfterFailedAcquire方法

/**
 * Checks and updates status for a node that failed to acquire.
 * Returns true if thread should block. This is the main signal
 * control in all acquire loops.  Requires that pred == node.prev.
 *
 * @param pred node's predecessor holding status
 * @param node the node
 * @return {@code true} if thread should block
 */
// pred是前置节点，Node是当前节点
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    // 获取前置节点的waitStatus
    int ws = pred.waitStatus;
    // SIGNAL的释义，请看上面的waitStatus状态值图示
    if (ws == Node.SIGNAL)
        /*
         * This node has already set status asking a release
         * to signal it, so it can safely park.
         */
        return true;
    if (ws > 0) { // 取消调度，cancel了
        /*
         * 
         * Predecessor was cancelled. Skip over predecessors and
         * indicate retry.
         */
        do {
            // 看下面的图示
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0); // 循环执行，直到waitStatus不大于0
        // 前置的next == 当前节点
        pred.next = node;
    } else {
        /*
         * 如果前驱正常，那就把前驱的状态设置为SIGNAL
         * waitStatus must be 0 or PROPAGATE.  Indicate that we
         * need a signal, but don't park yet.  Caller will need to
         * retry to make sure it cannot acquire before parking.
         */
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}
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
35
36
37
38
39
40
41
42
43

if (ws > 0) 图示

线程挂起阻塞

挂起线程

/**
     * Convenience method to park and then check if interrupted
     *
     * @return {@code true} if interrupted
     */
    private final boolean parkAndCheckInterrupt() {
        // 只有Unpark时才能解锁
        LockSupport.park(this);
        return Thread.interrupted();
    }
1
2
3
4
5
6
7
8
9
10

unlock 唤醒后继节点

那么lock获取锁的流程已经完事了，现在就是解锁的过程了

我们看看unLock();

public void unlock() {
        sync.release(1);
    }
1
2
3

public final boolean release(int arg) {
        if (tryRelease(arg)) {
            Node h = head;
            if (h != null && h.waitStatus != 0)
                // 接触被park的线程
                unparkSuccessor(h);
            return true;
        }
        return false;
    }
1
2
3
4
5
6
7
8
9
10

protected final boolean tryRelease(int releases) {
    		// 减去自旋增加的状态值
            int c = getState() - releases;
            if (Thread.currentThread() != getExclusiveOwnerThread())
                throw new IllegalMonitorStateException();
            boolean free = false;
            if (c == 0) { // 释放
                free = true;
                setExclusiveOwnerThread(null);
            }
   			// 状态最终需要设置回0
            setState(c);
            return free;
        }
1
2
3
4
5
6
7
8
9
10
11
12
13
14

真正的解锁，解除被挂起的线程，唤醒后继节点unparkSuccessor

/**
     * Wakes up node's successor, if one exists.
     *
     * @param node the node
     */
    private void unparkSuccessor(Node node) {
        /*
         * If status is negative (i.e., possibly needing signal) try
         * to clear in anticipation of signalling.  It is OK if this
         * fails or if status is changed by waiting thread.
         */
        int ws = node.waitStatus;
        if (ws < 0)
            compareAndSetWaitStatus(node, ws, 0);

        /*
         * Thread to unpark is held in successor, which is normally
         * just the next node.  But if cancelled or apparently null,
         * traverse backwards from tail to find the actual
         * non-cancelled successor.
         */
        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);
    }
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