【并发编程】第二章：从核心源码深入ReentrantLock锁

一：ReentrantLock简介

1.1 基本介绍

Java中常见的两种锁，一般就是synchronized和lock锁。ReentrantLock和synchronized一样都是互斥锁。ReentrantLock可以支持公平和非公平两种锁自定义，根据参数决定是公平锁还是非公平锁。

---

1.2 使用场景

1. 一般在锁竞争不激烈或者基本没有锁竞争的时候，推荐使用synchronized锁。在锁竞争比较激烈的时候推荐使用ReentrantLock锁，因为synchronized锁只有锁升级，当锁升级为重量级锁时无法降级为偏向锁，轻量级锁；如果竞争比较激烈会导致synchronized锁一直保持在重量级锁，而重量级锁会挂起线程效率低，即便后续过程竞争没有那么激烈了，synchronized还会保持在重量级锁。
2. ReentrantLock锁的使用对程序员的要求较高，程序员需要对其原理较为熟悉来能用的得心应手，否则可能会导致死锁等问题。
3. synchronized是关键字，而Lock是接口，因此ReentrantLock拓展出的功能更为强大更为完善，其API丰富，lock可以使用tryLock指定等待锁的时间，lock锁还提供了lockInterruptibly允许线程在获取锁的期间被中断等API。如果对ReentrantLock熟悉，更推荐使用ReentrantLock锁。

1.3 代码示例

public class Lock {
    //ReentrantLock锁的使用
    public void test1(){
        ReentrantLock reentrantLock=new ReentrantLock(true);
        //传入参数true表示公平锁,false表示非公平锁
        /*
        * lock锁必须要手动释放锁资源和synchronized锁自动释放不同
        *
        * 所以为了避免业务代码出现异常而导致手动释放锁代码无法执行最终造成死锁
        * 必须要把释放锁的代码写到finally代码块中
        * */
        reentrantLock.lock();

        try {
            /*
            * 执行代码
            * */
        }finally {
            reentrantLock.unlock();
        }

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

ReentrantLock支持锁重入

public class Lock {
    //ReentrantLock锁的使用
    public void test1(){
        ReentrantLock reentrantLock=new ReentrantLock(true);
        /*
        * lock锁必须要手动释放锁资源和synchronized锁自动释放不同
        *
        * 所以为了避免业务代码出现异常而导致手动释放锁代码无法执行最终造成死锁
        * 必须要把释放锁的代码写到finally代码块中
        * */
        reentrantLock.lock();

        try {
            /*
            * 执行代码
            * */
            //锁重入
            reentrantLock.lock();
            try {
                /*
                * 业务代码
                * */
            }finally {
                reentrantLock.unlock();
            }
            
        }finally {
            reentrantLock.unlock();
        }

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

---

二：ReentrantLock源码

2.1 ReentrantLock的Lock方法

公平

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

非公平

        final void lock() {
            if (compareAndSetState(0, 1))
                setExclusiveOwnerThread(Thread.currentThread());
            else
                acquire(1);
        }
1
2
3
4
5
6

如和判断线程获取锁资源成功？
ReentrantLock是基于AQS和CAS实现的，在AQS中有一个成员变量private volatile int state;这个成员变量我们可以理解为锁资源的状态，为0表示当前锁资源还没有被线程占用，大于0表示锁重入的次数。当前线程获取锁资源时会以CAS的方式将state由0->1，并通过setExclusiveOwnerThread（）设置占用锁资源的线程为当前线程。如果这两步均成功，代表线程获取锁资源成功。

公平和非公平Lock方法逻辑
公平锁：直接执行方法acquire(1);
非公平锁：因为是非公平，即便排队排在等待队列队头，新来的线程任然可以尝试去抢占锁资源，如果实在抢占不到锁资源就进入等待队列中，可能会挂起线程。

AQS的等待队列是基于那种数据结构实现的？
AQS中的等待队列中存放的是抢占不到锁资源的线程，在等待队列中的线程是可能会被挂起的，它是基于双向链表的数据结构来实现的。

---

2.2 ReentrantLock的acquire()方法

1. 执行tryAcquire(1),尝试获取锁资源（公平,非公平）,如果尝试获取锁资源成功，返回true，整个方法结束，如果没有获取到锁资源，则执行&&后的方法。
2. 执行addWaiter(),将当前线程封装成Node节点对象，并插入AQS双向链表的尾部。
3. 执行acquireQueued(),看当前线程所代表的节点对象是否在队列的首部(head.next),如果是则尝试获取到锁资源，如果不是则将当前线程所代表的节点对象挂起，进行阻塞。

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

---

2.3 ReentrantLock的tryAcquire()方法

公平锁

大致流程：
先获取锁的标识位state,如果state==0,则判断当前线程所代表的节点是否在队列的首部(head.next),是则尝试获取锁资源,不是则检查是否是锁重入，如果均不是则整个方法返回false。

        protected final boolean tryAcquire(int acquires) {
            final Thread current = Thread.currentThread();//获取当前线程
            int c = getState();//获取锁的标识位(数值可表示当前线程获取锁资源的次数)
            if (c == 0) {//锁资源没有被线程占用
             1. hasQueuedPredecessors():等待队列中有线程排队返回true，否则返回false
             2. &&后面的逻辑,当前线程尝试获取锁资源，并设置获取到锁资源的线程为当前线程。
                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

非公平锁
大致流程：
先获取锁的标识位state,如果state==0,直接尝试抢占锁资源;如果不是则检查是否是锁重入，如果均不是则整个方法返回false。

        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;
        }
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19

---

2.4 ReentrantLock的addWaiter()方法

addWaiter()就是将当前线程封装成一个节点，并插入到AQS双向链表的尾部。不区分公平和非公平。

    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;
            }
        }
        //上述插入过程可能会因为CAS并发失败而未执行
        //下面的enq(node)就是死循环直到node插入到链表尾部为止
        enq(node);
        return node;
    }
    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;
             }
         }
     }
 }
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

---

2.5 ReentrantLock的acquireQueued()方法

acquireQueued():方法就是判断当前线程所代表的节点能否有资格获取锁资源(即在AQS等待队列的头部),如果能则调用tryAcquire()尝试获取锁资源;否则将当前线程所代表的节点挂起。

    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;
                }
                1. shouldParkAfterFailedAcquire(p, node):做挂起当前节点前的预处理，设置当前节点之前的有效节点来唤醒当前节点。
                2. parkAndCheckInterrupt():挂起当前线程
                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

---

2.6 ReentrantLock的shouldParkAfterFailedAcquire()方法

节点的waitStatus标识位的取值

在ReentrantLock中这个waitStatus只有两种取值：CANCELLED 和SIGNAL。
CANCELLED：表示当前节点已经失效
SIGNAL：表示当前节点是有效的,可以负责唤醒后续的节点。

    private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        //获取当前节点前一个节点的ws
        int ws = pred.waitStatus;
        if (ws == Node.SIGNAL)//pred可以负责唤醒当前线程
            /*
             * This node has already set status asking a release
             * to signal it, so it can safely park.
             */
            return true;
        if (ws > 0) {//如果pred的状态是1，即已经失效
           //则遍历向前找，直到找到一个ws<=0的节点，让它负责唤醒当前线程
            /*
             * Predecessor was cancelled. Skip over predecessors and
             * indicate retry.
             */
            do {
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else {//如果是其它状态(-2,-3)，把ws设置成-1
            /*
             * 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);
        }
        //后面两种情况，双向链表发生的变动，当前节点可能因为前移而变成头节点，进而有尝试获取锁资源的资格，不必直接挂起线程。所以返回false，继续执行下一次循环，尝试获取锁资源。
        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

---

2.7 ReentrantLock的parkAndCheckInterrupt()方法

尝试挂起当前线程，并做相关的处理工作

    private final boolean parkAndCheckInterrupt() {
        //挂起当前线程
        LockSupport.park(this);
        return Thread.interrupted();
    }
1
2
3
4
5

---

2.7 ReentrantLock的unlock（）方法

释放锁资源

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

release()方法

    public final boolean release(int arg) {
        //尝试释放锁资源，返回true表示释放锁资源干净，即锁的标志位state=0
        //返回false表示还未释放干净
        if (tryRelease(arg)) {
            Node h = head;
            //h != null && h.waitStatus != 0 意思是如果链表中有节点，并且head的waitStatus=-1则唤醒head.next的线程
            if (h != null && h.waitStatus != 0)
                unparkSuccessor(h);
            return true;
        }
        return false;
    }
1
2
3
4
5
6
7
8
9
10
11
12

---

2.8 ReentrantLock的tryRelease()方法

        protected final boolean tryRelease(int releases) {
            int c = getState() - releases;//拿到state-1后的值
            //如果要释放锁资源的线程不是当前拥有锁资源的线程，直接抛异常
            if (Thread.currentThread() != getExclusiveOwnerThread())
                throw new IllegalMonitorStateException();
            boolean free = false;
            if (c == 0) {//c==0表示释放锁资源干净
                free = true;
                setExclusiveOwnerThread(null);
            }
            //更新state
            setState(c);
            return free;
        }
1
2
3
4
5
6
7
8
9
10
11
12
13
14