Java并发编程--Lock

Posted 2020-10-13 在周末

类结构图

 

 

Lock概述

　　Lock 实现提供了比使用 synchronized 方法和语句可获得的更广泛的锁定操作。synchronized方法或代码块的使用提供了对与每个对象相关的隐式监视器锁的访问，但却强制所有锁获取和释放均要出现在一个块结构中：当获取了多个锁时，它们必须以相反的顺序释放，且必须在与所有锁被获取时相同的词法范围内释放所有锁。Lock 实现提供了使用 synchronized 方法和语句所没有的其他功能，包括提供了一个非块结构的获取锁尝试 (tryLock())、一个获取可中断锁的尝试 (lockInterruptibly()) 和一个获取超时失效锁的尝试 (tryLock(long, TimeUnit))。

ReentrantLock

　　ReentrantLock是Lock的一种实现，它是一个可重入的互斥锁 ，它具有与使用 synchronized 方法和代码块所访问的隐式监视器锁相同的一些基本行为和语义，但功能更强大。ReentrantLock有公平锁和非公平锁之分，默认创建非公平锁。公平锁即谁先请求获取锁，谁就先获取到锁，即FIFO；反之则是非公平的。非公平锁的效率一般比公平锁的效率要高（为啥呢，可以减少线程切换），但公平锁能减少产生“饥饿”线程的概率，按需要选择。

　　同步状态（state）表示锁被一个线程重复获取的次数，这是支持可重入的关键。

　　使用示例

class X {
    private final ReentrantLock lock = new ReentrantLock();
    // ...
    public void m() {
        lock.lock();  // block until condition holds
        try {
            // ... method body
        } finally {
            lock.unlock()
        }
    }
}

　　　　注意：不要将获取锁的过程写在try块中，因为如果在获取锁（自定义锁的实现）时发生了异常，异常抛出的同时，也会导致锁无故释放。

　　实现原理

　　　　ReentrantLock怎么实现可重入、公平锁和非公平锁的呢？下面从源码角度研究它的实现原理。

　　　　获取锁和释放锁

　　　　　　调用 lock() 方法获取锁，ReentrantLock委托给同步器的lock()方法：sync.lock()。同样，调用unlock()方法释放锁，ReentrantLock也委托给同步器的release方法：sync.release(1);。

　　　　　　先看一下同步器sync的实现。 

 1 abstract static class Sync extends AbstractQueuedSynchronizer {
 2     private static final long serialVersionUID = -5179523762034025860L;
 3 
 4     abstract void lock();    //公平锁和非公平锁有不同的实现
 5     
 6     //非公平锁获取同步状态，NonfairSync的自定义方法tryAcquire()需要调用该方法
 7     final boolean nonfairTryAcquire(int acquires) {
 8         final Thread current = Thread.currentThread();
 9         int c = getState();
10         if (c == 0) {    //如果同步状态为0，表示锁没有被其他线程获取，则尝试CAS更新同步状态值，若CAS更新成功则获取锁成功
11             if (compareAndSetState(0, acquires)) {
12                 setExclusiveOwnerThread(current);
13                 return true;
14             }
15         }
16         else if (current == getExclusiveOwnerThread()) {    //可重入的关键，判断获取锁的线程是否是当前线程，如果是则同步状态+acquires，并返回true。
17             int nextc = c + acquires;
18             if (nextc < 0) // overflow
19                 throw new Error("Maximum lock count exceeded");
20             setState(nextc);
21             return true;
22         }
23         return false;
24     }
25     
26     //释放锁
27     protected final boolean tryRelease(int releases) {
28         int c = getState() - releases;
29         if (Thread.currentThread() != getExclusiveOwnerThread())    //判断当前线程是否持有锁
30             throw new IllegalMonitorStateException();
31         boolean free = false;
32         if (c == 0) {    //针对可重入的设计，如果该锁被获取了n次，那么需要释放n次，c才等于0，这是才能返回true。
33             free = true;
34             setExclusiveOwnerThread(null);
35         }
36         setState(c);
37         return free;
38     }
39     //判断当前线程是否持有锁
40     protected final boolean isHeldExclusively() {
41         return getExclusiveOwnerThread() == Thread.currentThread();
42     }
43     //构造此lock的Condition，实现等待/通知方式的线程间通信
44     final ConditionObject newCondition() {
45         return new ConditionObject();
46     }
47 }

　　　　　　由同步器源码可以看出，Sync继承了AQS，是一个ReentrantLock类的抽象内部类，它重写了释放锁tryRelease方法。由于获取锁的机制在公平锁和非公平锁中实现不同，所以tryAcquire方法的重写放在了Sync的子类FairSync和NonfairSync中，抽象方法lock()在FairSync和NonfairSync中的实现不一样。

　　　　　　先看非公平锁

 1 //非公平同步器
 2 static final class NonfairSync extends Sync {
 3     private static final long serialVersionUID = 7316153563782823691L;
 4 
 5     /**
 6      * Performs lock.  Try immediate barge, backing up to normal
 7      * acquire on failure.
 8      */
 9     final void lock() {
10         if (compareAndSetState(0, 1))    //首先尝试CAS更新同步状态，更新成功即获取锁，否则调用AQS的模板方法acquire
11             setExclusiveOwnerThread(Thread.currentThread());
12         else
13             acquire(1);    //调用AQS的独占式获取同步状态方法
14     }
15     
16     //自定义重写的tryAcquire方法，返回true表示成功，false表示失败。AQS的acquire方法会调用该方法
17     protected final boolean tryAcquire(int acquires) {
18         return nonfairTryAcquire(acquires);
19     }
20 }

　　　　　　再看公平锁

 1 //公平同步器
 2 static final class FairSync extends Sync {
 3     private static final long serialVersionUID = -3000897897090466540L;
 4     //重写Sync中的抽象方法
 5     final void lock() {
 6         acquire(1);    //调用AQS的独占式获取同步状态方法
 7     }
 8 
 9     /**
10      * Fair version of tryAcquire.  Don\'t grant access unless
11      * recursive call or no waiters or is first.
12      */
13     //自定义重写的tryAcquire方法
14     protected final boolean tryAcquire(int acquires) {
15         final Thread current = Thread.currentThread();
16         int c = getState();
17         if (c == 0) {
18             //与非公平锁不同的是，此处多了hasQueuedPredecessors()判断，该方法是实现公平锁的关键。
19             //如果hasQueuedPredecessors返回true，表示有其他线程先于当前线程等待获取锁，此时为了实现公平，保证等待时间最长的线程先获取到锁，不能执行CAS。
20             //CAS可能会破坏公平性。反之，如果hasQueuedPredecessors返回false,则可以执行CAS更新同步状态尝试获取锁。
21             if (!hasQueuedPredecessors() &&
22                 compareAndSetState(0, acquires)) {
23                 setExclusiveOwnerThread(current);
24                 return true;
25             }
26         }
27         else if (current == getExclusiveOwnerThread()) {
28             //当前线程已经获取到锁，重入
29             int nextc = c + acquires;
30             if (nextc < 0)
31                 throw new Error("Maximum lock count exceeded");
32             setState(nextc);
33             return true;
34         }
35         return false;
36     }
37 }

　　　　　　和非公平同步器不同的是，公平同步器在进行CAS更新同步状态之前会调用 hasQueuedPredecessors() 方法判断是否有其他线程先于当前线程等待获取锁，然后再根据判断结果执行相应的操作。那么 hasQueuedPredecessors() 方法是怎么实现的呢？hasQueuedPredecessors()是AQS提供的一个模板方法，具体实现见源码。

 1 //是否有其他线程先于当前线程等待获取锁，即判断队列中是否有前驱节点
 2 public final boolean hasQueuedPredecessors() {
 3     // The correctness of this depends on head being initialized
 4     // before tail and on head.next being accurate if the current
 5     // thread is first in queue.
 6     Node t = tail; // Read fields in reverse initialization order
 7     Node h = head;
 8     Node s;
 9     return h != t &&
10         ((s = h.next) == null || s.thread != Thread.currentThread());
11     //1)如果h==t成立，队列为空，无前驱节点，返回false。
12     //2）如果h!=t成立，从head节点的next是否为null，如果为null，返回true。什么情况下h!=t的同时h.next==null？？，
　　　　//有其他线程第一次正在入队时，可能会出现。见AQS的enq方法，compareAndSetHead(node)完成，还没执行tail = head语句时，此时tail=null,head=newNode,head.next-null。
13     //3）如果h!=t成立，从head节点的next是否不为null，则判断是否是当前线程，如果是返回false，否则有前驱节点，返回true
14 }

　　　　对比公平锁和非公平锁：在非公平锁中，由于刚释放锁的线程再次获取同步状态的几率会非常大，所以使得其他线程只能在同步队列中等待。公平性锁保证了锁的获取按照FIFO原则，而代价是进行大量的线程切换。非公平性锁虽然可能造成线程“饥饿”，但极少的线程切换，保证了其更大的吞吐量。

　　　　响应中断获取锁

1 //调用AQS的响应中断获取同步状态方法
2 public void lockInterruptibly() throws InterruptedException {
3     sync.acquireInterruptibly(1);
4 }

 

　　　　超时获取锁

1 //调用AQS的超时获取同步状态方法
2 public boolean tryLock(long timeout, TimeUnit unit)
3     throws InterruptedException {
4     return sync.tryAcquireNanos(1, unit.toNanos(timeout));
5 }

 

　　ReentrantLock与synchronized监视器锁的比较

　　　　1）所有 synchronized 能做的，Lock都能做，它拥有与 synchronized 相同的内存和并发性语义，还拥有 synchronized 所没有的特性，例如超时获取锁、可中断获取、无块结构锁、多个条件变量或者轮询锁。

　　　　2）synchronized的同步机制是JVM实现的，获取锁和释放锁等是隐式的，不易出错；lock是通过代码实现的，获取锁和释放锁等是显式的，程序员很容易忘记用 finally 块释放锁，这对程序非常有害。

　　　　3）当 JVM 用 synchronized 管理锁定请求和释放时，JVM 在生成线程转储时能够包括锁定信息。这些对调试非常有价值，因为它们能标识死锁或者其他异常行为的来源。 Lock 类只是普通的类，JVM 不知道具体哪个线程拥有 Lock 对象。

　　　　4）在资源竞争不是很激烈的情况下，Synchronized的性能要优于ReetrantLock，但是在资源竞争很激烈的情况下，Synchronized的性能会下降几十倍，但是ReetrantLock的性能能维持常态；

　　如何选择？

　　　　在确实需要一些 synchronized 所没有的特性的时候，比如时间锁等候、可中断锁等候、无块结构锁、多个条件变量或者轮询锁。 ReentrantLock 还具有可伸缩性的好处，应当在高度争用的情况下使用它，但要前提是保证替换后性能更好。

Condition（等待/通知）

　　类似于synchronized同步关键字与监视器方法（定义在java.lang.Object上的wait()、wait(long timeout)、notify()以及notifyAll()方法）实现等待/通知模式，Condition和Lock配合也可以实现等待/通知模式，而且功能更强大。

　　Condition的功能

　　　　Condition对象是Lock对象创建出来的，在调用Condition中的所有方法前，必须获取到该Condition对应的锁。

        //当前线程等待直到被通知或被中断
        void await() throws InterruptedException;
        //当前线程等待直到被通知,不响应中断
        void awaitUninterruptibly();                
        //当前线程等待直到被通知或被中断或超过指定的时间（纳秒），返回值表示剩余的时间
        long awaitNanos(long nanosTimeout) throws InterruptedException;    
        //当前线程等待直到被通知或被中断或超过指定的时间，可指定时间单位
        boolean await(long time, TimeUnit unit) throws InterruptedException;
        //当前线程等待直到被通知或被中断或超过截止时间，没到截止时间就被唤醒，返回true，否则返回false
        boolean awaitUntil(Date deadline) throws InterruptedException;    
        void signal();        //唤醒一个等待的线程，该线程从等待方法返回前必须重新获取锁，
        void signalAll();    //唤醒所有等待的线程，每一个线程必须重新尝试获取锁，能够从等待方法返回的线程必须获得锁

 

　　　　当前线程调用等待方法，并释放与此Condition相关的锁。唤醒方式有以下四种：

　　　　　　1）其他某个线程调用此 Condition 的 signal() 方法，并且碰巧将当前线程选为被唤醒的线程；

　　　　　　2）其他某个线程调用此 Condition 的 signalAll() 方法；

　　　　　　3）其他某个线程中断（调用interrupt方法）当前线程，且支持中断线程的挂起；

　　　　　　4）发生“虚假唤醒”。

　　　　在所有情况下，当前线程从等待方法返回之前，都必须重新获取与此Condition对应的锁。在线程await返回时，可以保证它已经获取了Condition对应的锁。

　　Condition的实现原理

　　　　ConditionObject是接口Condition的实现类，且是AQS的内部类。下面以ConditionObject的源码来分析实现原理。

　　　　Condition队列（等待队列）

　　　　　　ConditionObject内部维护了一个FIFO的等待队列，用于存储在该Condition上等待的线程引用。Node节点就是AQS的内部类，即同步队列使用的Node类。当线程调用等待方法，该线程就会构造线程节点并从尾部加入等待队列。

1 public class ConditionObject implements Condition, java.io.Serializable {
2     /** First node of condition queue. */
3     private transient Node firstWaiter;    //队列的first节点
4     /** Last node of condition queue. */
5     private transient Node lastWaiter;    //队列的last节点
6 }

　　　　await（等待）

　　　　　　先看一下 await() 方法的源码

 1 //如果当前线程被中断，将抛出InterruptedException
 2 public final void await() throws InterruptedException {
 3     if (Thread.interrupted())
 4         throw new InterruptedException();
 5     //将当前线程构造Node并加入等待队列尾部
 6     Node node = addConditionWaiter();
 7     //释放同步状态
 8     int savedState = fullyRelease(node);
 9     int interruptMode = 0;
10     while (!isOnSyncQueue(node)) {    //在阻塞线程前，需要判断该节点是否在同步队列中，如果在则没必要阻塞当前线程，否则阻塞当前线程
11         LockSupport.park(this);        //阻塞当前线程，当其他线程调用signal时，如果该线程被选中会被唤醒，此时该线程节点已经被移到同步队列中，isOnSyncQueue(node)返回true，会退出While循环。
12         if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
13             break;
14     }
15     //通过acquireQueued方法加入到获取同步状态的竞争中，acquireQueued自旋获取排他锁。获取锁成功，则从await中退出；获取取锁失败则阻塞在acquireQueued方法中。这就保证了从await方法返回的线程可能获取了锁。
16     //如果线程在wait的过程中被中断过，那么acquireQueued方法返回true，否则返回false
17     //THROW_IE=-1表示线程在从wait退出时将抛InterruptedException异常
18     if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
19         interruptMode = REINTERRUPT;    //REINTERRUPT=1，表示在线程在从wait退出时reinterrupt
20     if (node.nextWaiter != null) // clean up if cancelled
21         unlinkCancelledWaiters();
22     if (interruptMode != 0)
23         reportInterruptAfterWait(interruptMode);    //对异常的处理
24 }
25 
26 //将当前线程构造Node并加入等待队列尾部，此时还未释放锁，所以不必使用CAS更新，锁可以保证线程安全
27 private Node addConditionWaiter() {
28     Node t = lastWaiter;
29     // If lastWaiter is cancelled, clean out.
30     if (t != null && t.waitStatus != Node.CONDITION) {
31         unlinkCancelledWaiters();
32         t = lastWaiter;
33     }
34     //waitStatus == Node.CONDITION 表明线程正在Condition上等待
35     Node node = new Node(Thread.currentThread(), Node.CONDITION);
36     if (t == null)
37         firstWaiter = node;
38     else
39         t.nextWaiter = node;
40     lastWaiter = node;
41     return node;
42 }

　　　　　　由源码可看出，await() 方法是响应中断的，如果当前线程被其他线程中断，则抛出 InterruptedException 异常。await() 方法可以分为以下几步：

　　　　　　　　1）将当前线程构造成Node并加入到该Condition的等待队列的尾部，addConditionWaiter()方法；

　　　　　　　　2）释放同步状态，唤醒该锁的同步队列中后继节点线程，fullyRelease()方法；

　　　　　　　　3）阻塞当前线程，LockSupport.park(this)，等待被唤醒。

　　　　　　其中释放同步状态方法 fullyRelease() 是 AQS 提供的一个方法，看源码：

 1 //释放同步状态，如果释放失败，会抛IllegalMonitorStateException且node的waitStatus被置为CANCELLED。
 2 final int fullyRelease(Node node) {
 3     boolean failed = true;
 4     try {
 5         int savedState = getState();    //获取当前同步状态
 6         if (release(savedState)) {    //release(arg)是熟悉的面孔，AQS提供的独占式释放同步状态的方法
 7             failed = false;
 8             return savedState;    //返回同步状态值，包括重入的次数
 9         } else {
10             throw new IllegalMonitorStateException();
11         }
12     } finally {
13         if (failed)
14             node.waitStatus = Node.CANCELLED;
15     }
16 }
17 
18 //独占式释放同步状态，
19 public final boolean release(int arg) {
20     if (tryRelease(arg)) {    //自定义独占式tryRelease方法
21         Node h = head;
22         if (h != null && h.waitStatus != 0)
23             unparkSuccessor(h);    //唤醒后继节点
24         return true;
25     }
26     return false;
27 }

　　　　如果从队列（同步队列和等待队列）的角度看await()方法，当调用await()方法时，相当于同步队列的首节点（获取了锁的节点）移动到Condition的等待队列中。同步队列的首节点并不会直接加入等待队列，而是通过addConditionWaiter()方法把当前线程构造成一个新的节点并将其加入等待队列中。

　　　　signal（通知）

　　　　　　先看signal()方法的源码

 1 //通知
 2 public final void signal() {
 3     if (!isHeldExclusively())    //执行signal方法的前提是获取了锁
 4         throw new IllegalMonitorStateException();
 5     Node first = firstWaiter;
 6     if (first != null)
 7         doSignal(first);    //通知等待队列中的firstNode，等待时间最长的节点
 8 }
 9 
10 private void doSignal(Node first) {
11     do {
12         if ( (firstWaiter = first.nextWaiter) == null)    //首节点的后继节点为null，在唤醒了首节点后，等待队列会变为空
13             lastWaiter = null;
14         first.nextWaiter = null;
15     } while (!transferForSignal(first) &&
16              (first = firstWaiter) != null);
17 }

　　　　　　由源码可看出，首先检查当前线程是否获取了锁，如果没有获取锁，抛IllegalMonitorStateException异常，然后将等待队列中的首节点移动到同步队列中。

　　　　　　其中 transferForSignal(Node node) 方法是将一个等待队列中的节点移动到同步队列中的关键，它是 AQS 提供的方法。

 1 //AQS的方法，将一个等待队列中的节点移动到同步队列中，如果成功则返回true。
 2 final boolean transferForSignal(Node node) {
 3     /*
 4      * If cannot change waitStatus, the node has been cancelled.
 5      */
 6     //将节点的waitStatus由CONDITION改为0（初始化状态），条件队列中的节点只有两种状态，CONDITION和cancelled
 7     if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
 8         return false;
 9 
10     /*
11      * Splice onto queue and try to set waitStatus of predecessor to
12      * indicate that thread is (probably) waiting. If cancelled or
13      * attempt to set waitStatus fails, wake up to resync (in which
14      * case the waitStatus can be transiently and harmlessly wrong).
15      */
16     Node p = enq(node);    //入队，返回node的前驱（即node入队之前的tail节点）
17     int ws = p.waitStatus;
18     //如果node的前驱节点p被取消（ws > 0），或者cas设置p的WaitStatus为SIGNAL失败
19     if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
20         LockSupport.unpark(node.thread);    //唤醒node节点的线程，被唤醒的线程会从await方法的循环中退出，（isOnSyncQueue(Node node)方法返回true，节点已经在同步队列中。
21     return true;
22 }

　　　　　　与await()方法中把当前线程构造成一个新的节点并将其加入等待队列不同的是，signal()方法中直接把等待队列中的节点Node转移到同步队列中。

　　　　　　调用LockSupport.unpark(node.thread)方法唤醒当前线程，此时，阻塞线程被唤醒，开始执行await()方法的第12行。现在再来看 isOnSyncQueue(node) 方法就比较清晰了，由于此时节点已经在同步队列中了，所以该方法返回true，退出while循环。然后通过acquireQueued方法加入到获取同步状态的竞争中，这样就保证了从await方法返回的线程肯定获取了锁。因为在acquireQueued方法中，通过自旋阻塞机制获取锁，如果acquireQueued返回结果，则获取锁成功，结束await()方法；获取取锁失败则阻塞在acquireQueued方法中。

　　　　signalAll（通知全部）

　　　　　　与signal的区别就是将条件队列中所有node都转移到同步队列中。

 1 public final void signalAll() {
 2     if (!isHeldExclusively())
 3         throw new IllegalMonitorStateException();
 4     Node first = firstWaiter;
 5     if (first != null)
 6         doSignalAll(first);
 7 }
 8 //将条件队列中所有node都转移到同步队列中
 9 private void doSignalAll(Node first) {
10     lastWaiter = firstWaiter = null;
11     do {
12         Node next = first.nextWaiter;
13         first.nextWaiter = null;
14         transferForSignal(first);
15         first = next;
16     } while (first != null);
17 }

 　　Condition 与 wait/notify 的比较

	wait/notify	Condition
前置条件	获取对象的锁	也需要获取对象的锁
等待队列个数	一个	多个
是否支持等待不响应中断	不支持	支持
超时等待状态	支持	支持
等待状态到将来某个时间	不支持	支持
唤醒等待队列中的一个线程	支持	支持
唤醒等待队列中的所有线程	支持	支持

 

 

参考资料

　　《Java 理论与实践》 https://www.ibm.com/developerworks/cn/java/j-jtp10264/index.html

　　《Java并发编程的艺术》　　

　　（ReentrantLock实现原理深入探究）http://www.cnblogs.com/xrq730/p/4979021.html