哪个网站教人做美食/深圳专门做seo的公司

在 五：AbstractQueuedSynchronizer 文章中，我们介绍了 AQS 的基本原理。 ReentrantLock 是我们比较常用的一种锁，也是基于 AQS 实现的。所以，接下来我们就来分析一下 ReentrantLock 锁的实现

1、ReentrantLock 入门

ReentrantLock：可重入且互斥的锁。

案例：

public class Test {private static final Lock lock = new ReentrantLock();public static void test() {// 获取锁lock.lock();try {System.out.println(Thread.currentThread().getName() + "获取到锁了");//业务代码,使用部分花费100毫秒Thread.sleep(100);} catch (Exception e) {e.printStackTrace();} finally {lock.unlock();System.out.println(Thread.currentThread().getName() + "释放了锁");}}public static void main(String[] args){Runnable task = Test3::test;new Thread(task, "thread1").start();new Thread(task, "thread2").start();}
}

运行结果如下：

thread1获取到锁了
thread1释放了锁
thread2获取到锁了
thread2释放了锁

效果和 synchronized 的一样，线程1获取到锁了，线程2需要等待线程1释放锁后才可以获取锁

2、ReentrantLock 源码解析

类图如下：

ReentrantLock 有三个内部类：

• Sync：继承 AbstractQueuedSynchronizer（AQS），同步队列器
• NonfairSync：非公平锁
• FairSync：公平锁

2.1、构造方法：默认为非公平锁

public ReentrantLock() {sync = new NonfairSync();
}
// 带有参数的构造方法：公平、非公平
public ReentrantLock(boolean fair) {sync = fair ? new FairSync() : new NonfairSync();
}

2.2、三大内部类

查看三大内部类用到的方法及方法调用的关系：

abstract static class Sync extends AbstractQueuedSynchronizer {// 抽象方法：由公平锁、非公平锁 两种方式实现abstract void lock();// 用于非公平方式，尝试获取锁final boolean nonfairTryAcquire(int acquires) {//...}// 实现了 AQS 的方法protected boolean tryRelease(int arg) {throw new UnsupportedOperationException();}
}// 非公平锁
static final class NonfairSync extends Sync {final void lock() {//...}protected final boolean tryAcquire(int acquires) {//...}
}// 公平锁
static final class FairSync extends Sync {final void lock() {//...}protected final boolean tryAcquire(int acquires) {//...}
}

2.2、lock()：加锁【不可中断锁】

由于默认的是非公平锁的加锁，所以我们来分析下非公平锁是如何加锁的

public void lock() {sync.lock();
}

调用 Sync#lock() 方法：是一个抽象方法，由【公平锁】、【非公平锁】子类去实现：

abstract static class Sync extends AbstractQueuedSynchronizer {abstract void lock();
}

非公平锁：

static final class NonfairSync extends Sync {final void lock() {if (compareAndSetState(0, 1)) {setExclusiveOwnerThread(Thread.currentThread());} else {acquire(1);}}
}

这个方法有两步：

1. 使用 CAS 来获取 state 资源，如果成功设置 1，代表 state 资源获取锁成功，此时记录下当前占用 state 的线程 setExclusiveOwnerThread(Thread.currentThread());
2. 如果获取锁失败，则执行 acquire(1) 方法

2.2.1、acquire() 方法 —— AQS【模板方法】

这个方法是由 AQS 提供的模板方法，AQS#acquire() 方法：

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

acquire() 方法：先尝试获取锁，如果获取锁成功，则返回；否则，调用 acquireQueued() 方法，将线程添加到 CLH 同步等待队列中

如图：

2.2.2.1 tryAcquire() 方法 —— AQS，由子类去实现

此方法在 AQS 中是一个空方法，留个子类自己去实现。上面我们使用的是非公平锁。所以回到 NonfairSync#tryAcquire() 方法：

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

调用 Sync#nonfairTryAcquire() 方法：

final boolean nonfairTryAcquire(int acquires) {final Thread current = Thread.currentThread();int c = getState();// 未加锁if (c == 0) {// 加锁：CAS 操作把 state 赋值为 1，exclusiveOwnerThread() 赋值为 currentThread，然后返回 trueif (compareAndSetState(0, acquires)) {setExclusiveOwnerThread(current);return true;}} else if (current == getExclusiveOwnerThread()) {// 已加锁，并且是持锁线程是当前线程（可重入），计数器加1int nextc = c + acquires;if (nextc < 0) {// 溢出throw new Error("Maximum lock count exceeded");        }setState(nextc);return true;}return false;
}

tryAcquire() 方法：尝试加锁

1. 先判断当前锁是否已经被释放：如果已经被释放，那么，通过 CAS 操作去加锁，如果加锁成功，则直接返回 true【非公平锁：因为队列中的线程与新线程都可以 CAS 获取锁啊，新来的线程不需要排队】
2. 如果锁还未被释放，那么判断当前线程是否是持有锁的线程，那么，就将计数器加 1，并返回 true【可重入锁】。因为就是当前线程持有锁，所以可以使用 setState() 【未使用 CAS】去更新 state 值
3. 否则，返回 false，加锁失败

2.2.2.2. addWaiter() 方法 —— AQS

如果尝试加锁失败【tryAcquire() 方法返回 false】， 则执行 acquireQueued() 方法，将线程添加到 CLH 同步等待队列中

private Node addWaiter(Node mode) {// 构造一个Node对象，参数是当前线程、mode 对象，mode 表示该节点的共享/排他性，值为null为排他模式，不为null则共享模式Node node = new Node(Thread.currentThread(), mode);// 拿到队列尾节点Node pred = tail;// 尝试快速方式将新 node 直接放到队尾。// 如果尾节点不为空if (pred != null) {// 先把新加入的节点的前驱节点设置为尾节点，新加入的节点会加入队列的尾部node.prev = pred;// 通过 CAS 操作把新节点设置为尾节点，传入原来的尾节点 pred 和新节点 node 做判断，保证并发安全if (compareAndSetTail(pred, node)) {// 把新节点设置为原来尾节点的后继节点pred.next = node;// 返回新节点，这个节点里封装了当前的线程return node;}}// CAS 添加到队尾（会初始化队头）enq(node);return node;
}

addWaiter() 方法：将包含有当前线程的 Node 节点【独占模式】入队，并返回这个 Node

1. 如果尾结点存在，则采用 CAS 的方式将当前线程入队
2. 尾结点为空则执行 enq() 方法

enq() 方法：

private Node enq(final Node node) {// 这是一个死循环，不满足一定的条件就不会跳出循环for (;;) {// 获取队列尾节点Node t = tail;// 如果为 null，其实这是个循环判断，可能下次再做判断时，就有其他线程已经往队列中添加了节点，那么tail尾节点可能就不为空了，就走else逻辑了if (t == null) {// 必须初始化队头// 新建一个Node对象，通过 CAS 设置成头节点，这个 head 其实是冗余节点if (compareAndSetHead(new Node()))tail = head;} else {// 将此 node 添加到队尾// 尾节点不为空，则把尾节点设置为新节点的前驱节点node.prev = t;// CAS操作，把新节点设置为尾节点if (compareAndSetTail(t, node)) {// CAS 成功后，则把新节点设置为原来尾节点的后继节点t.next = node;return t;}}}
}

通过死循环的方式，来保证节点的正确添加，可以发现只有当新节点被设置为尾节点时，方法才能返回，然后再配合上 CAS，节点一个一个的被加到队列中，一个一个的接着被设置为尾节点，并发的操作，串行的感觉

使用 CAS 创建 head 节点的时候只是简单调用了 new Node() 方法，并不像其他节点那样记录 thread，这是为什么？

因为 head 结点为虚结点，它只代表持有锁线程占用了 state，至于占用 state 的是哪个线程，其实是调用了上文的 setExclusiveOwnerThread(current) ，即记录在 exclusiveOwnerThread 属性里

2.2.2.3 acquireQueued() 方法 —— AQS

final boolean acquireQueued(final Node node, int arg) {boolean failed = true;try {// 标记等待过程中是否被中断过boolean interrupted = false;// 自旋for (;;) {// 获取 node 的前驱节点final Node p = node.predecessor();// 如果 node 的前驱是头节点，那么便有资格去尝试获取资源（可能是上一个获取到锁的节点释放完资源唤醒自己的，当然也可能被 interrupt 了）if (p == head && tryAcquire(arg)) {// 那么把当前节点设置为头节点，同时把当前节点的前驱节点置为nullsetHead(node);// 再把前头节点p的后继节点设置为 null，这样前头节点就没有任何引用了，帮助 GC，清理前头节点p.next = null;// 成功获取资源failed = false;// 返回等待过程中是否被中断过return interrupted;}// 当 node 的前驱节点不是头节点或者获取锁失败时，判断是否需要阻塞等待，如果需要等待，那么就调用parkAndCheckInterrupt()方法阻塞等待if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) {// 如果等待过程中被中断过，哪怕只有那么一次，就将interrupted标记为trueinterrupted = true;}}} finally {if (failed) {// 如果等待过程中没有成功获取资源（如timeout）| 抛异常，那么取消结点在队列中的等待。cancelAcquire(node);        }}
}

acquireQueued() 方法：每个节点都会进行 2 次自旋【for(;;) {代码快}】，每次自旋，如果前驱节点是 head 节点的话，就会去尝试加锁，如果加锁失败，就会调用 shouldParkAfterFailedAcquire() 方法去掉 CANCEL 状态的 节点，并且修改前驱节点的 waitStatues 为 SINGAL。第二次自旋时，会调用 parkAndCheckInterrupt() 方法将当前节点阻塞起来

1. 如果当前节点的前驱是 head 节点，那么调用 tryAcquire() 方法尝试加锁（有可能此时持有锁的线程已经释放了锁），如果加锁成功，那么，通过 CAS 操作把当前节点设置为 head 节点
2. 如果当前节点的前驱不是 head 节点 | 尝试加锁失败，则调用 shouldParkAfterFailedAcquire() 方法，判断锁是否应该停止自旋进入阻塞状态；如果需要进入阻塞状态，则调用 parkAndCheckInterrupt() 方法进行阻塞

为什么要先自旋 2 次，再进行阻塞？而不是直接就阻塞呢？

马上阻塞意味着线程从运行态转为阻塞态 ，涉及到用户态向内核态的切换，而且唤醒后也要从内核态转为用户态，开销相对比较大，所以 AQS 对这种入队的线程采用的方式是让它们先自旋来竞争锁；

如果当前锁是独占锁，如果锁一直被被持有锁线程占有， 其它线程 一直自旋没太大意义，反而会占用 CPU，影响性能，所以更合适的方式是让它们自旋一两次，竞争不到锁后识趣地阻塞起来，以等待前置节点释放锁后再来唤醒它

如果锁在自旋过程中被中断了，或者自旋超时了，应该处于「取消」状态

setHead() 方法：

private void setHead(Node node) {head = node;node.thread = null;node.prev = null;
}

将 head 设置成当前结点后，要把节点的 thread, pre 设置成 null，因为之前分析过了：head 是虚节点，不保存除 waitStatus（结点状态）的其他信息，所以这里把 thread ,pre 置为空，因为占有锁的线程由 exclusiveThread 记录了，如果 head 再记录 thread 不仅多此一举，反而在释放锁的时候要多操作一遍 head 的 thread 释放

2.2.2.3.1、shouldParkAfterFailedAcquire() 方法 —— AQS

waitStatus：默认是 0，可取值为：

• CANCELLED：1。线程已被取消，这种状态节点会被忽略，并移除队列等待 GC
• SIGNAL：-1。线程被挂起，后继节点可以尝试抢占锁
• CONDITION：-2。线程正在等待某些条件
• PROPAGATE：-3。共享模式下，无条件所有等待线程尝试抢占锁

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {// 获取前驱状态int ws = pred.waitStatus;if (ws == Node.SIGNAL) {// 表示要阻塞return true;}// 如果状态大于0，表示前驱节点需要做的请求被取消了if (ws > 0) {// 如果前驱放弃了，那就一直往前找，直到找到最近一个正常等待的状态，并排在它的后边// 注意：那些放弃的结点，由于被自己“加塞”到它们前边，它们相当于形成一个无引用链，稍后就会被 GC 回收！do {node.prev = pred = pred.prev;} while (pred.waitStatus > 0);pred.next = node;} else {// 如果前驱正常，那就通过 CAS 操作把前驱的状态设置成 SIGNALcompareAndSetWaitStatus(pred, ws, Node.SIGNAL);}return false;
}

shouldParkAfterFailedAcquire() 方法：判断锁是否应该停止自旋（并进入阻塞状态）

1. 判断当前节点的前驱节点的 waitStatus 是否为 SIGNAL，如果是，则返回 true，表示当前节点停止自旋，应该进入阻塞状态；
2. 如果不是且前驱节点的 waitStatus 为 CANCELLED 状态，那么就把当前节点放到 waitStatus 为非 CANCELLED 的节点后面
3. 否则，将前驱节点的 waitStatus 置为 SIGNAL 状态

2.2.2.3.2、 parkAndCheckInterrupt() 方法 —— AQS

private final boolean parkAndCheckInterrupt() {// 调用park()使线程进入waiting状态LockSupport.park(this);// 如果被唤醒，查看自己是不是被中断的。return Thread.interrupted();
}

shouldParkAfterFailedAcquire() 方法返回 true，代表线程可以进入阻塞中断。

为什么要判断线程是否中断过呢？

因为如果线程在阻塞期间收到了中断，唤醒（转为运行态）获取锁后（acquireQueued() 为 true）需要补一个中断

static void selfInterrupt() {Thread.currentThread().interrupt();
}

由于在整个抢锁过程中，我们都是不响应中断的。那如果在抢锁的过程中发生了中断怎么办呢？总不能假装没看见呀。AQS 的做法简单的记录有没有有发生过中断，如果返回的时候发现曾经发生过中断，则在退出 acquire() 方法之前，就调用 selfInterrupt() 方法自我中断一下，就好像将这个发生在抢锁过程中的中断“推迟”到抢锁结束以后再发生一样。

2.2.2.3.3、cancelAcquire() 方法 —— AQS

private void cancelAcquire(Node node) {// 如果节点为空，直接返回if (node == null) {return;}node.thread = null;// 跳过所有取消状态的结点Node pred = node.prev;while (pred.waitStatus > 0) {node.prev = pred = pred.prev;}Node predNext = pred.next;node.waitStatus = Node.CANCELLED;// 如果当前取消结点为尾结点，使用 CAS 则将尾结点设置为其前驱节点，如果设置成功，则尾结点的 next 指针设置为空if (node == tail && compareAndSetTail(node, pred)) {compareAndSetNext(pred, predNext, null);} else {// 如果当前节点取消了，那是不是要把当前节点的前驱节点指向当前节点的后继节点，但是我们之前也说了，要唤醒或阻塞结点，须在其前驱节点的状态为 SIGNAL 的条件才能操作，所以在设置 pre 的 next 节点时要保证 pre 结点的状态为 SIGNALint ws;if (pred != head &&((ws = pred.waitStatus) == Node.SIGNAL ||(ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&pred.thread != null) {Node next = node.next;if (next != null && next.waitStatus <= 0)compareAndSetNext(pred, predNext, next);} else {// 如果 pre 为 head，或者  pre 的状态设置 SIGNAL 失败，则直接唤醒后继结点去竞争锁，之前我们说过， SIGNAL 的结点取消（或释放锁）后可以唤醒后继结点unparkSuccessor(node);}node.next = node; // help GC}
}

什么时候会出现 CANCEL 状态的节点？

1. 线程发生中断
2. 线程获取锁超时

2.3、unlock() 方法

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

调用 AQS#release() 方法，是个模板方法

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

如果锁释放成功，则唤醒同步等待队列中 head 的一个后继节点，并返回 true；否则，直接返回 false

2.3.1、tryRelease() 方法 —— Sync

protected final boolean tryRelease(int releases) {int c = getState() - releases;// 判断持有锁的线程是不是当前线程if (Thread.currentThread() != getExclusiveOwnerThread()) {throw new IllegalMonitorStateException();}boolean free = false;// 如果 state==0，证明此次锁释放成功if (c == 0) {free = true;setExclusiveOwnerThread(null);}setState(c);return free;
}

如果不是持有锁线程释放锁，则抛异常；如果是，且 state 为 0，则返回 true，表示释放锁成功；否则，返回 false

锁释放成功后就应该唤醒 head 节点之后的节点来竞争锁

为什么释放锁的条件为啥是 h != null && h.waitStatus != 0 ？

1. h == null：head 节点为 null，有两种可能
   1. 当前只有一个线程访问，且就是持有锁的线程，即：同步等待队列中没有阻塞线程
   2. 其它线程正在运行竞争锁，只是还未初始化 head 节点，既然其它线程正在运行，也就无需执行唤醒操作
2. h != null && h.waitStatus == 0：head 的后继节点 T 正在自旋竞争锁（T 还未将它的前驱节点 head 的状态修改为 SIGNAL），无需唤醒
3. h != null && h.waitStatus < 0：此时 waitStatus 值可能为 SIGNAL，或 PROPAGATE，这两种情况说明后继结点阻塞需要被唤醒

2.3.2、unparkSuccessor() 方法 —— AQS

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 为空，或者其为取消状态，说明 s 是无效节点，此时需要执行 for 里的逻辑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);}
}

unparkSuccessor() 方法：在同步等待队列中，从尾向前获取最后一个非取消状态的结点，并将其唤醒

2.4、公平锁 & 非公平锁

公平锁与非公平锁的实现区别：tryAcquire() 方法

hasQueuedPredecessors() 方法：判断当前线程前面有没有在排队的线程，有则返回 true，否则返回 false

public final boolean hasQueuedPredecessors() {Node t = tail; Node h = head;Node s;return h != t && ((s = h.next) == null || s.thread != Thread.currentThread());
}

区别如下：

• 公平锁：多个线程按照申请锁的顺序去获得锁。当多个线程进行访问时，如果同步等待队列中有线程等待【锁已经被某个线程持有】，那么它不会去尝试获取锁，而是直接进入队列去排队
  • 优点：所有的线程都能得到资源，不会饿死在队列中
  • 缺点：吐量会下降很多，队列里面除了第一个线程，其他的线程都会阻塞，CPU 唤醒阻塞线程的开销会很大
• 非公平锁：当多个线程进行访问时，即使锁已经被持有，且同步等待队列中已有其它线程在等待，那么，它也会先去尝试获取锁。如果能获取锁，则就加锁；否则，进入同步等待队列中，先自旋再阻塞
  • 优点：可以减少 CPU 唤醒线程的开销，整体的吞吐效率会高点，CPU 也不必取唤醒所有线程，会减少唤起线程的数量
  • 缺点：这样可能导致队列中间的线程一直获取不到锁或者长时间获取不到锁，导致饿死

2.5、lockInterruptibly() 方法 —— 加锁【响应中断】

public void lockInterruptibly() throws InterruptedException {sync.acquireInterruptibly(1);
}

2.5.1、acquireInterruptibly() 方法 —— AQS

public final void acquireInterruptibly(int arg) throws InterruptedException {// 如果线程中断，则抛异常if (Thread.interrupted()) {throw new InterruptedException();}if (!tryAcquire(arg)) {doAcquireInterruptibly(arg);}
}

doAcquireInterruptibly() 方法：发生中断后，会将此节点置为 CANCEL 状态

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;failed = false;return;}// 线程被唤醒时，如果发生中断，则抛异常if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) {throw new InterruptedException();}   }} finally {if (failed)cancelAcquire(node);}
}

• lockInterruptibly()：发生了中断是会直接抛 InterruptedException 异常，响应中断，立即停止获取锁的流程【你的线程对中断敏感】【可以用来解决死锁问题】
• lock()：发生了中断之后，会继续尝试获取锁，通过返回中断标识延迟中断【你的线程对中断不敏感，当然，也要注意处理中断】

2.6、tryLock() 方法 —— 尝试获取锁

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) {throw new Error("Maximum lock count exceeded");}setState(nextc);return true;}return false;
}

tryLock() 方法：尝试获取锁，获取锁成功，返回 true；否则，返回 false

2.7、boolean tryLock(long time, TimeUnit unit) 方法 —— 超时获取锁

static final long spinForTimeoutThreshold = 1000L;private boolean doAcquireNanos(int arg, long nanosTimeout) throws InterruptedException {if (nanosTimeout <= 0L) {return false;}final long deadline = System.nanoTime() + nanosTimeout;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;failed = false;return true;}nanosTimeout = deadline - System.nanoTime();if (nanosTimeout <= 0L) {return false;}// 阻塞给定的超时时间if (shouldParkAfterFailedAcquire(p, node) && nanosTimeout > spinForTimeoutThreshold) {LockSupport.parkNanos(this, nanosTimeout);}// 中断抛异常if (Thread.interrupted()) {throw new InterruptedException();}}} finally {if (failed)cancelAcquire(node);}
}

3、ReentrantLock & synchronized

虽然在性能上 ReentrantLock 和 synchronized 没有什么区别，但 ReentrantLock 相比 synchronized 而言功能更加丰富，使用起来更为灵活，也更适合复杂的并发场景。比如：公平锁/非公平锁、尝试获取锁、超时获取锁以及中断等待锁的线程等

共同点：

1. 都是独占锁、非公平锁
2. 都是可重入的

不同点：

1. 底层实现不同：synchronized 是 JVM层面的锁，是Java关键字，通过 monitor 对象来完成（monitorenter 与 monitorexit），ReentrantLock 是 JDK 的 API
2. 锁的对象不同：synchronzied 锁的是对象，锁是保存在对象头里面的，根据对象头数据来标识是否有线程获得锁/争抢锁；ReentrantLock 是根据 volatile 变量 state 标识锁的获得/争抢
3. 实现机制不同：synchronized 的实现涉及到锁的升级，具体为无锁、偏向锁、自旋锁、向内核态申请重量级锁；ReentrantLock 实现则是通过利用 CAS（CompareAndSwap）自旋机制保证线程操作的原子性和 volatile 保证数据可见性以实现锁的功能
4. 释放锁方式上不同：synchronized 不需要用户去手动释放锁，synchronized 代码执行完后系统会自动释放锁；ReentrantLock需要用户去手动释放锁，如果没有手动释放锁，就可能导致死锁现象。一般通过 lock() 和 unlock() 方法配合 try/finally 语句块来完成