2. AQS

2.1 AQS原理

2.1.1 概述

全称是 AbstractQueuedSynchronizer，是阻塞式锁和相关的同步器工具的框架

2.2.2 特点

  • 用 state 属性来表示资源的状态（分独占模式和共享模式），子类需要定义如何维护这个状态，控制如何获取锁和释放锁
    • getState - 获取 state 状态
    • setState - 设置 state 状态
    • compareAndSetState - cas 机制设置 state 状态
    • 独占模式是只有一个线程能够访问资源，而共享模式可以允许多个线程访问资源
  • 提供了基于 FIFO 的等待队列，类似于 Monitor 的 EntryList
  • 条件变量来实现等待、唤醒机制，支持多个条件变量，类似于 Monitor 的 WaitSet

子类主要实现这样一些方法（默认抛出 UnsupportedOperationException）

  • tryAcquire
  • tryRelease
  • tryAcquireShared
  • tryReleaseShared
  • isHeldExclusively ```java // 获取锁的姿势 // 如果获取锁失败 if (!tryAcquire(arg)) { // 入队, 可以选择阻塞当前线程 park unpark }

// 释放锁的姿势 // 如果释放锁成功 if (tryRelease(arg)) { // 让阻塞线程恢复运行 }

  1. <a name="nycyv"></a>
  2. ### 2.2.3 自定义锁
  3. 下面实现一个不可重入的阻塞式锁：使用 AbstractQueuedSynchronizer 自定义一个同步器来实现自定义锁，代码如下：
  4. ```java
  5. @Slf4j(topic = "c.Test2")
  6. public class Test2 {
  8.     public static void main(String[] args) {
  9.         MyLock lock = new MyLock();
  11.         new Thread(()->{
  12.             lock.lock();
  13.             try {
  14.                 Thread.sleep(1000);
  15.                 log.debug("locking...");
  16.             } catch (InterruptedException e) {
  17.                 e.printStackTrace();
  18.             } finally {
  19.                 lock.unlock();
  20.                 log.debug("unlocking....");
  21.             }
  22.         }, "t1").start();
  24.         new Thread(()->{
  25.             lock.lock();
  26.             try {
  27.                 log.debug("locking...");
  28.             } finally {
  29.                 lock.unlock();
  30.                 log.debug("unlocking....");
  31.             }
  32.         }, "t2").start();
  33.     }
  34. }
  36. // 自定义锁，不可重复锁
  37. class MyLock implements Lock {
  39.     // 独占锁 ,同步器类
  40.     class MySync extends AbstractQueuedSynchronizer{
  41.         @Override // 尝试获得锁 0无锁 1加上锁
  42.         protected boolean tryAcquire(int arg) {
  43.             if (compareAndSetState(0 ,1)){
  44.                 // 加上了锁，并设置owner为当前线程
  45.                 setExclusiveOwnerThread(Thread.currentThread());
  46.             }
  47.             return true;
  48.         }
  50.         @Override //尝试释放锁
  51.         protected boolean tryRelease(int arg) {
  52.             setExclusiveOwnerThread(null);
  53.             setState(0); // 这一步要在后面，state是volatile的保证前面属性的可见性
  54.             return true;
  55.         }
  57.         @Override  //是否持有独占锁
  58.         protected boolean isHeldExclusively() {
  59.             return getState() == 1;
  60.         }
  62.         public Condition newCondition(){
  63.             return new ConditionObject();
  64.         }
  65.     }
  67.     private MySync mySync = new MySync();
  69.     @Override   //加锁（不成功进入等待队列）
  70.     public void lock() {
  71.         mySync.acquire(1);
  72.     }
  74.     @Override   // 加锁， 可打断
  75.     public void lockInterruptibly() throws InterruptedException {
  76.         mySync.acquireInterruptibly(1);
  77.     }
  79.     @Override  // 尝试加锁，一次
  80.     public boolean tryLock() {
  81.         return mySync.tryAcquire(1);
  82.     }
  84.     @Override // 尝试加锁，带超时
  85.     public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
  86.         return mySync.tryAcquireNanos(1, unit.toNanos(time));
  87.     }
  89.     @Override // 解锁
  90.     public void unlock() {
  91.         mySync.release(1);
  92.     }
  94.     @Override // 创建条件变量
  95.     public Condition newCondition() {
  96.         return mySync.newCondition();
  97.     }
  1. 12:34:21.191 [t2] DEBUG c.Test2 - locking...
  2. 12:34:21.193 [t2] DEBUG c.Test2 - unlocking....
  3. 12:34:22.189 [t1] DEBUG c.Test2 - locking...
  4. 12:34:22.189 [t1] DEBUG c.Test2 - unlocking....

2.2 ReentrantLock原理

ReentrantLock 提供了抽象的Sync（继承了AQS），并实现了两个同步器，实现公平锁（FairSync）和非公平锁（NonfairSync），默认是非公平锁，实现了Lock接口
image.png

2.2.1 构造器

带参数和不带参数的构造器，不带参默认为非公平的

  1.  public ReentrantLock() {
  2.         sync = new NonfairSync();
  3.     }
  1.  public ReentrantLock(boolean fair) {
  2.         sync = fair ? new FairSync() : new NonfairSync();
  3.     }

2.2.2 加锁过程

NonfairSync 继承自 Sync 继承自AQS

  1. static final class NonfairSync extends Sync {
  2.         private static final long serialVersionUID = 7316153563782823691L;
  4.         /**
  5.          * Performs lock.  Try immediate barge, backing up to normal
  6.          * acquire on failure.
  7.          */
  8.         final void lock() {
  9.             if (compareAndSetState(0, 1))
  10.                 setExclusiveOwnerThread(Thread.currentThread());
  11.             else
  12.                 acquire(1);
  13.         }
  15.         protected final boolean tryAcquire(int acquires) {
  16.             return nonfairTryAcquire(acquires);
  17.         }
  18.     }

没有竞争时：
image.png
有竞争时：调用acquire方法

  1. public final void acquire(int arg) {
  2.     // 没有重新加锁成功 且 将其加入到等待队列
  3.         if (!tryAcquire(arg) &&
  4.             acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
  5.             selfInterrupt();
  6.     }

image.png

  • lock方法中CAS 尝试将 state 由 0 改为 1，结果失败lock方法中进一步调用acquire方法，进入 tryAcquire 逻辑，这里我们认为这时 state 已经是1，结果仍然失败
  • 接下来进入 acquire方法的addWaiter 逻辑，构造 Node 队列
    • 图中黄色三角表示该 Node 的 waitStatus 状态，其中 0 为默认正常状态
    • Node 的创建是懒惰的
    • 其中第一个 Node 称为 Dummy（哑元）或哨兵，用来占位，并不关联线程

image.png

acquireQueued方法：
当前线程进入 acquire方法的 acquireQueued 逻辑

  • acquireQueued 会在一个死循环中不断尝试获得锁，失败后进入 park 阻塞
  • 如果自己是紧邻着 head（排第二位），那么再次 tryAcquire 尝试获取锁，我们这里设置这时 state 仍为 1，失败
  • 进入 shouldParkAfterFailedAcquire 逻辑，将前驱 node，即 head 的 waitStatus 改为 -1，这次返回 false（意思就是你尝试了好几次都没获得锁，就进入阻塞，让前面那个节点唤醒你）

image.png

  • shouldParkAfterFailedAcquire 执行完毕回到 acquireQueued ，再次 tryAcquire 尝试获取锁，当然这时 state 仍为 1，失败
  • 当再次进入 shouldParkAfterFailedAcquire 时，这时因为其前驱 node 的 waitStatus 已经是 -1，这次返回 true
  • 进入 parkAndCheckInterrupt， Thread-1 park（灰色表示已经阻塞）

image.png
再从次有多个线程经历上述过程竞争失败，变成这个样子7 共享模型之AQS - 图7
源码：

  1. // Sync 继承自 AQS
  2. static final class NonfairSync extends Sync {
  3.     private static final long serialVersionUID = 7316153563782823691L;
  5.      // 加锁实现
  6.     final void lock() {
  7.         // 首先用 cas 尝试（仅尝试一次）将 state 从 0 改为 1, 如果成功表示获得了独占锁
  8.         if (compareAndSetState(0, 1))
  9.             setExclusiveOwnerThread(Thread.currentThread());
  10.         else
  11.             // 如果尝试失败，进入 ㈠
  12.             acquire(1);
  13.     }
  15.     // ㈠ AQS 继承过来的方法, 方便阅读, 放在此处 尝试加锁失败的过程
  16.     public final void acquire(int arg) {
  17.         // ㈡ tryAcquire
  18.         if (
  19.                 // 再次尝试获取锁    
  20.                 !tryAcquire(arg) &&
  21.                 // 当 tryAcquire 返回为 false 时, 先调用 addWaiter ㈣, 接着 acquireQueued ㈤
  22.                  acquireQueued(addWaiter(Node.EXCLUSIVE), arg)
  23.         ) {
  24.             selfInterrupt();
  25.         }
  26.     }
  28.     // ㈡ 进入 ㈢
  29.     protected final boolean tryAcquire(int acquires) {
  30.         return nonfairTryAcquire(acquires);
  31.     }
  33.     // ㈢ Sync 继承过来的方法, 方便阅读, 放在此处
  34.     final boolean nonfairTryAcquire(int acquires) {
  35.         final Thread current = Thread.currentThread();
  36.         int c = getState();
  37.         // 如果还没有获得锁
  38.         if (c == 0) {
  39.             // 尝试用 cas 获得, 这里体现了非公平性: 不去检查 AQS 队列
  40.             if (compareAndSetState(0, acquires)) {
  41.                 setExclusiveOwnerThread(current);
  42.                 return true;
  43.             }
  44.         }
  45.         // 如果已经获得了锁, 线程还是当前线程, 表示发生了锁重入
  46.         else if (current == getExclusiveOwnerThread()) {
  47.             // state++
  48.             int nextc = c + acquires;
  49.             if (nextc < 0) // overflow
  50.                 throw new Error("Maximum lock count exceeded");
  51.             setState(nextc);
  52.             return true;
  53.         }
  54.         // 获取失败, 回到调用处
  55.         return false;
  56.     }
  58.     // ㈣ AQS 继承过来的方法, 方便阅读, 放在此处  将该节点放到等待队列中
  59.     private Node addWaiter(Node mode) {
  60. // 将当前线程关联到一个 Node 对象上, 模式为独占模式，新建的Node的waitstatus默认为0，因为waitstatus是成员变量，默认被初始化为0
  61.         Node node = new Node(Thread.currentThread(), mode);
  62.         // 如果 tail 不为 null, cas 尝试将 Node 对象加入 AQS 队列尾部
  63.         Node pred = tail;
  64.         if (pred != null) {
  65.             node.prev = pred;
  66.             if (compareAndSetTail(pred, node)) {
  67.                 // 双向链表
  68.                 pred.next = node;
  69.                 return node;
  70.             }
  71.         }
  72.         //如果tail为null，尝试将 Node 加入 AQS, 进入 ㈥
  73.         enq(node);
  74.         return node;
  75.     }
  77.     // ㈥ AQS 继承过来的方法, 方便阅读, 放在此处
  78.     private Node enq(final Node node) {
  79.         for (;;) {
  80.             Node t = tail;
  81.             if (t == null) {
  82.                 // 还没有, 设置 head 为哨兵节点（不对应线程，状态为 0）
  83.                 if (compareAndSetHead(new Node())) {
  84.                     tail = head;
  85.                 }
  86.             } else {
  87.                 // cas 尝试将 Node 对象加入 AQS 队列尾部
  88.                 node.prev = t;
  89.                 if (compareAndSetTail(t, node)) {
  90.                     t.next = node;
  91.                     return t;
  92.                 }
  93.             }
  94.         }
  95.     }
  97.     // ㈤ AQS 继承过来的方法, 方便阅读, 放在此处   对等待队列进行处理
  98.     final boolean acquireQueued(final Node node, int arg) {
  99.         boolean failed = true;
  100.         try {
  101.             boolean interrupted = false;
  102.             for (;;) {
  103.                 final Node p = node.predecessor();
  104.                 // 上一个节点是 head, 表示轮到自己（当前线程对应的 node）了, 尝试获取
  105.                 if (p == head && tryAcquire(arg)) {
  106.                     // 获取成功, 设置自己（当前线程对应的 node）为 head
  107.                     setHead(node);
  108.                     // 上一个节点 help GC
  109.                     p.next = null;
  110.                     failed = false;
  111.                     // 返回中断标记 false
  112.                     return interrupted;
  113.                 }
  114.                 if (
  115.                     // 判断是否应当 park, 进入 ㈦
  116.                     shouldParkAfterFailedAcquire(p, node) &&
  117.                     // park 等待, 此时 Node 的状态被置为 Node.SIGNAL ㈧
  118.                     parkAndCheckInterrupt()
  119.                 ) {
  120.                     interrupted = true;
  121.                 }
  122.             }
  123.         } finally {
  124.             if (failed)
  125.                 cancelAcquire(node);
  126.         }
  127.     }
  129.     // ㈦ AQS 继承过来的方法, 方便阅读, 放在此处  是否需要在获取失败时被阻塞
  130.     private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
  131.         // 获取上一个节点的状态
  132.         int ws = pred.waitStatus;
  133.         if (ws == Node.SIGNAL) {
  134.             // 上一个节点都在阻塞, 那么自己也阻塞好了
  135.             return true;
  136.         }
  137.         // > 0 表示取消状态
  138.         if (ws > 0) {
  139.             // 上一个节点取消, 那么重构删除前面所有取消的节点, 返回到外层循环重试
  140.             do {
  141.                 node.prev = pred = pred.prev;
  142.             } while (pred.waitStatus > 0);
  143.             pred.next = node;
  144.         } else {
  145.             // 这次还没有阻塞
  146.             // 但下次如果重试不成功, 则需要阻塞，这时需要设置上一个节点状态为 Node.SIGNAL
  147.             compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
  148.         }
  149.         return false;
  150.     }
  152.     // ㈧ 阻塞当前线程
  153.     private final boolean parkAndCheckInterrupt() {
  154.         LockSupport.park(this);
  155.         return Thread.interrupted();
  156.     }
  157. }

2.2.3 释放锁流程

Thread-0 释放锁，进入 tryRelease 流程，如果成功

  • 设置 exclusiveOwnerThread 为 null
  • state = 0

image.png
如果当前队列不为 null，并且 head 的 waitStatus = -1，进入 unparkSuccessor 流程：

  • unparkSuccessor 中会找到队列中离 head 最近的一个 Node（没取消的），unpark 恢复其运行，本例中即为 Thread-1
  • 回到 Thread-1 的 acquireQueued 流程

image.png
如果加锁成功（没有竞争），会设置 （acquireQueued 方法中）

  • exclusiveOwnerThread 为 Thread-1，state = 1
  • head 指向刚刚 Thread-1 所在的 Node，该 Node 清空 Thread
  • 原本的 head 因为从链表断开，而可被垃圾回收

如果这时候有其它线程来竞争（非公平的体现），例如这时有 Thread-4 来了
image.png

  1. Thread-4 被设置为 exclusiveOwnerThread，state = 1

  2. Thread-1 再次进入 acquireQueued 流程，获取锁失败，重新进入 park 阻塞 ```java // Sync 继承自 AQS static final class NonfairSync extends Sync { // 解锁实现 public void unlock() {

    1.  sync.release(1);

    }

    // AQS 继承过来的方法, 方便阅读, 放在此处 public final boolean release(int arg) {

    1.  // 尝试释放锁, 进入 ㈠
    2.  if (tryRelease(arg)) {
    3.      // 队列头节点 unpark
    4.      Node h = head;
    5.      if (
    6.          // 队列不为 null
    7.          h != null &&
    8.          // waitStatus == Node.SIGNAL 才需要 unpark
    9.          h.waitStatus != 0
    10.      ) {
    11.          // unpark AQS 中等待的线程, 进入 ㈡
    12.          unparkSuccessor(h);
    13.      }
    14.      return true;
    15.  }
    16.  return false;

    }

    // ㈠ Sync 继承过来的方法, 方便阅读, 放在此处 protected final boolean tryRelease(int releases) {

    1.  // state--
    2.  int c = getState() - releases;
    3.  if (Thread.currentThread() != getExclusiveOwnerThread())
    4.      throw new IllegalMonitorStateException();
    5.  boolean free = false;
    6.  // 支持锁重入, 只有 state 减为 0, 才释放成功
    7.  if (c == 0) {
    8.      free = true;
    9.      setExclusiveOwnerThread(null);
    10.  }
    11.  setState(c);
    12.  return free;

    }

    // ㈡ AQS 继承过来的方法, 方便阅读, 放在此处 private void unparkSuccessor(Node node) {

    1.  // 如果状态为 Node.SIGNAL 尝试重置状态为 0, 如果线程获取到了锁那么后来头结点会被抛弃掉
    2.  // 不成功也可以
    3.  int ws = node.waitStatus;
    4.  if (ws < 0) {
    5.      compareAndSetWaitStatus(node, ws, 0);
    6.  }
    7.  // 找到需要 unpark 的节点, 但本节点从 AQS 队列中脱离, 是由唤醒节点完成的
    8.  Node s = node.next;
    9.  // 不考虑已取消的节点, 从 AQS 队列从后至前找到队列最前面需要 unpark 的节点
    10.  if (s == null || s.waitStatus > 0) {
    11.      s = null;
    12.      for (Node t = tail; t != null && t != node; t = t.prev)
    13.          if (t.waitStatus <= 0)
    14.              s = t;
    15.  }
    16.  if (s != null)
    17.      LockSupport.unpark(s.thread);

    } }

  1. <a name="hkJVf"></a>
  2. ### 2.2.4 锁重入原理
  3. 如果是同一个线程，则将state++；释放后，会将state--。如果state为0，才会释放锁。
  4. ```java
  5. static final class NonfairSync extends Sync {
  6.     // ...
  8.     // Sync 继承过来的方法, 方便阅读, 放在此处  不公平的获取锁的方式
  9.     final boolean nonfairTryAcquire(int acquires) {
  10.         final Thread current = Thread.currentThread();
  11.         int c = getState();
  12.         if (c == 0) {
  13.             // 如果c为0，表示没有线程加锁，那么该线程进行加锁
  14.             if (compareAndSetState(0, acquires)) {
  15.                 // 设置锁头信息 owner
  16.                 setExclusiveOwnerThread(current);
  17.                 return true;
  18.             }
  19.         }
  20.         // 如果已经获得了锁, 线程还是当前线程, 表示发生了锁重入，更新state
  21.         else if (current == getExclusiveOwnerThread()) {
  22.             // state++
  23.             int nextc = c + acquires;
  24.             if (nextc < 0) // overflow
  25.                 throw new Error("Maximum lock count exceeded");
  26.             setState(nextc);
  27.             return true;
  28.         }
  29.         return false;
  30.     }
  32.     // Sync 继承过来的方法, 方便阅读, 放在此处 释放锁
  33.     protected final boolean tryRelease(int releases) {
  34.         // state--
  35.         int c = getState() - releases;
  36.         if (Thread.currentThread() != getExclusiveOwnerThread())
  37.             throw new IllegalMonitorStateException();
  38.         // c不为0，还是当前线程的锁，返回释放失败。
  39.         boolean free = false;
  40.         // 支持锁重入, 只有 state 减为 0, 才释放成功
  41.         if (c == 0) {
  42.             // 返回释放成功，并将owner置为null
  43.             free = true;
  44.             setExclusiveOwnerThread(null);
  45.         }
  46.         setState(c);
  47.         return free;
  48.     }
  49. }

2.2.5 可打断原理

  • 不可打断模式：

在此模式下，即使它被打断，仍会驻留在 AQS 队列中，一直要等到获得锁后方能得知自己被打断了。

  1. // Sync 继承自 AQS
  2. static final class NonfairSync extends Sync {
  3.     // ...
  5.     private final boolean parkAndCheckInterrupt() {
  6.         // 如果打断标记已经是 true, 则 park 会失效
  7.         LockSupport.park(this);
  8.         // interrupted 会清除打断标记
  9.         return Thread.interrupted();
  10.     }
  12.     final boolean acquireQueued(final Node node, int arg) {
  13.         boolean failed = true;
  14.         try {
  15.             boolean interrupted = false;
  16.             for (;;) {
  17.                 final Node p = node.predecessor();
  18.                 if (p == head && tryAcquire(arg)) {
  19.                     setHead(node);
  20.                     p.next = null;
  21.                     failed = false;
  22.                     // 还是需要获得锁后, 才能返回打断状态
  23.                     return interrupted;
  24.                 }
  25.                 if (
  26.                         shouldParkAfterFailedAcquire(p, node) &&
  27.                                 parkAndCheckInterrupt()
  28.                 ) {
  29.                     // 如果是因为 interrupt 被唤醒, 返回打断状态为 true
  30.                     interrupted = true;
  31.                 }
  32.             }
  33.         } finally {
  34.             if (failed)
  35.                 cancelAcquire(node);
  36.         }
  37.     }
  39.     public final void acquire(int arg) {
  40.         if (
  41.                 !tryAcquire(arg) &&
  42.                         acquireQueued(addWaiter(Node.EXCLUSIVE), arg)
  43.         ) {
  44.             // 如果打断状态为 true
  45.             selfInterrupt();
  46.         }
  47.     }
  49.     static void selfInterrupt() {
  50.         // 重新产生一次中断，这时候线程是如果正常运行的状态，那么不是出于sleep等状态，interrupt方法就不会报错
  51.         Thread.currentThread().interrupt();
  52.     }
  53. }
  54. }

  • 可打断模式： ```java static final class NonfairSync extends Sync { public final void acquireInterruptibly(int arg) throws InterruptedException {

    1.   if (Thread.interrupted())
    2.       throw new InterruptedException();
    3.   // 如果没有获得到锁, 进入 ㈠
    4.   if (!tryAcquire(arg))
    5.       doAcquireInterruptibly(arg);

    }

    // ㈠ 可打断的获取锁流程 private void doAcquireInterruptibly(int arg) throws InterruptedException {

    1.   final Node node = addWaiter(Node.EXCLUSIVE);
    2.   boolean failed = true;
    3.   try {
    4.       for (;;) {
    5.           final Node p = node.predecessor();
    6.           if (p == head && tryAcquire(arg)) {
    7.               setHead(node);
    8.               p.next = null; // help GC
    9.               failed = false;
    10.               return;
    11.           }
    12.           if (shouldParkAfterFailedAcquire(p, node) &&
    13.                   parkAndCheckInterrupt()) {
    14.               // 在 park 过程中如果被 interrupt 会进入此
    15.               // 这时候抛出异常, 而不会再次进入 for (;;)
    16.               throw new InterruptedException();
    17.           }
    18.       }
    19.   } finally {
    20.       if (failed)
    21.           cancelAcquire(node);
    22.   }

    } }

  1. 区别在于不可打断模式是将打断标记置为true，再次进入for循环（在AQS队列中）；可打断模式直接抛出异常，不会进入for循环。
  2. <a name="qzarC"></a>
  3. ### 2.2.6 公平锁原理
  4. 非公平锁主要区别在于 tryAcquire 方法的实现：在加锁之前会判断AQS队列有没有前驱结点
  5. ```java
  6. static final class FairSync extends Sync {
  7.     private static final long serialVersionUID = -3000897897090466540L;
  8.     final void lock() {
  9.         acquire(1);
  10.     }
  12.     // AQS 继承过来的方法, 方便阅读, 放在此处
  13.     public final void acquire(int arg) {
  14.         if (
  15.                 !tryAcquire(arg) &&
  16.                         acquireQueued(addWaiter(Node.EXCLUSIVE), arg)
  17.         ) {
  18.             selfInterrupt();
  19.         }
  20.     }
  21.     // 与非公平锁主要区别在于 tryAcquire 方法的实现
  22.     protected final boolean tryAcquire(int acquires) {
  23.         final Thread current = Thread.currentThread();
  24.         int c = getState();
  25.         if (c == 0) {
  26.             // 先检查 AQS 队列中是否有前驱节点, 没有才去竞争
  27.             if (!hasQueuedPredecessors() &&
  28.                     compareAndSetState(0, acquires)) {
  29.                 setExclusiveOwnerThread(current);
  30.                 return true;
  31.             }
  32.         }
  33.         else if (current == getExclusiveOwnerThread()) {
  34.             int nextc = c + acquires;
  35.             if (nextc < 0)
  36.                 throw new Error("Maximum lock count exceeded");
  37.             setState(nextc);
  38.             return true;
  39.         }
  40.         return false;
  41.     }
  43.     // ㈠ AQS 继承过来的方法, 方便阅读, 放在此处
  44.     public final boolean hasQueuedPredecessors() {
  45.         Node t = tail;
  46.         Node h = head;
  47.         Node s;
  48.         // h != t 时表示队列中有 Node
  49.         return h != t &&
  50.                 (
  51.                         // (s = h.next) == null 表示队列中还有没有老二
  52.                         (s = h.next) == null || // 或者队列中老二线程不是此线程
  53.                                 s.thread != Thread.currentThread()
  54.                 );
  55.     }
  56. }

2.2.7 条件变量实现原理

每个条件变量其实就对应着一个等待队列，其实现类是 ConditionObject

  • await流程

开始 Thread-0 持有锁，调用 await，进入 ConditionObject 的 addConditionWaiter 流程 创建新的Node 状态为 -2（Node.CONDITION），关联 Thread-0，加入等待队列尾部7 共享模型之AQS - 图11
接下来进入 AQS 的 fullyRelease 流程，释放同步器上的锁
image.png
unpark AQS 队列中的下一个节点，竞争锁，假设没有其他竞争线程，那么 Thread-1 竞争成功
image.png
park 阻塞 Thread-0
image.png

  • signal流程

假设 Thread-1 要来唤醒 Thread-0
image.png
进入 ConditionObject 的 doSignal 流程，取得等待队列中第一个 Node，即 Thread-0 所在 Node
image.png
执行 transferForSignal 流程，将该 Node 加入 AQS 队列尾部，将 Thread-0 的 waitStatus 改为 0，Thread-3 的waitStatus 改为 -1
image.png
Thread-1 释放锁，进入 unlock 流程。

  1. public class ConditionObject implements Condition, java.io.Serializable {
  2.     private static final long serialVersionUID = 1173984872572414699L;
  4.     // 第一个等待节点
  5.     private transient Node firstWaiter;
  7.     // 最后一个等待节点
  8.     private transient Node lastWaiter;
  9.     public ConditionObject() { }
  10.     // ㈠ 添加一个 Node 至等待队列
  11.     private Node addConditionWaiter() {
  12.         Node t = lastWaiter;
  13.         // 所有已取消的 Node 从队列链表删除, 见 ㈡
  14.         if (t != null && t.waitStatus != Node.CONDITION) {
  15.             unlinkCancelledWaiters();
  16.             t = lastWaiter;
  17.         }
  18.         // 创建一个关联当前线程的新 Node, 添加至队列尾部
  19.         Node node = new Node(Thread.currentThread(), Node.CONDITION);
  20.         if (t == null)
  21.             firstWaiter = node;
  22.         else
  23.             t.nextWaiter = node;
  24.         lastWaiter = node;
  25.         return node;
  26.     }
  27.     // 唤醒 - 将没取消的第一个节点转移至 AQS 队列
  28.     private void doSignal(Node first) {
  29.         do {
  30.             // 已经是尾节点了
  31.             if ( (firstWaiter = first.nextWaiter) == null) {
  32.                 lastWaiter = null;
  33.             }
  34.             first.nextWaiter = null;
  35.         } while (
  36.             // 将等待队列中的 Node 转移至 AQS 队列, 不成功且还有节点则继续循环 ㈢
  37.                 !transferForSignal(first) &&
  38.                         // 队列还有节点
  39.                         (first = firstWaiter) != null
  40.         );
  41.     }
  43.     // 外部类方法, 方便阅读, 放在此处
  44.     // ㈢ 如果节点状态是取消, 返回 false 表示转移失败, 否则转移成功
  45.     final boolean transferForSignal(Node node) {
  46.         // 设置当前node状态为0（因为处在队列末尾），如果状态已经不是 Node.CONDITION, 说明被取消了
  47.         if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
  48.             return false;
  49.         // 加入 AQS 队列尾部
  50.         Node p = enq(node);
  51.         int ws = p.waitStatus;
  52.         if (
  53.             // 插入节点的上一个节点被取消
  54.                 ws > 0 ||
  55.                         // 插入节点的上一个节点不能设置状态为 Node.SIGNAL
  56.                         !compareAndSetWaitStatus(p, ws, Node.SIGNAL)
  57.         ) {
  58.             // unpark 取消阻塞, 让线程重新同步状态
  59.             LockSupport.unpark(node.thread);
  60.         }
  61.         return true;
  62.     }
  63. // 全部唤醒 - 等待队列的所有节点转移至 AQS 队列
  64. private void doSignalAll(Node first) {
  65.     lastWaiter = firstWaiter = null;
  66.     do {
  67.         Node next = first.nextWaiter;
  68.         first.nextWaiter = null;
  69.         transferForSignal(first);
  70.         first = next;
  71.     } while (first != null);
  72. }
  74.     // ㈡
  75.     private void unlinkCancelledWaiters() {
  76.         // ...
  77.     }
  78.     // 唤醒 - 必须持有锁才能唤醒, 因此 doSignal 内无需考虑加锁
  79.     public final void signal() {
  80.         // 如果没有持有锁，会抛出异常
  81.         if (!isHeldExclusively())
  82.             throw new IllegalMonitorStateException();
  83.         Node first = firstWaiter;
  84.         if (first != null)
  85.             doSignal(first);
  86.     }
  87.     // 全部唤醒 - 必须持有锁才能唤醒, 因此 doSignalAll 内无需考虑加锁
  88.     public final void signalAll() {
  89.         if (!isHeldExclusively())
  90.             throw new IllegalMonitorStateException();
  91.         Node first = firstWaiter;
  92.         if (first != null)
  93.             doSignalAll(first);
  94.     }
  95.     // 不可打断等待 - 直到被唤醒
  96.     public final void awaitUninterruptibly() {
  97.         // 添加一个 Node 至等待队列, 见 ㈠
  98.         Node node = addConditionWaiter();
  99.         // 释放节点持有的锁, 见 ㈣
  100.         int savedState = fullyRelease(node);
  101.         boolean interrupted = false;
  102.         // 如果该节点还没有转移至 AQS 队列, 阻塞
  103.         while (!isOnSyncQueue(node)) {
  104.             // park 阻塞
  105.             LockSupport.park(this);
  106.             // 如果被打断, 仅设置打断状态
  107.             if (Thread.interrupted())
  108.                 interrupted = true;
  109.         }
  110.         // 唤醒后, 尝试竞争锁, 如果失败进入 AQS 队列
  111.         if (acquireQueued(node, savedState) || interrupted)
  112.             selfInterrupt();
  113.     }
  114.     // 外部类方法, 方便阅读, 放在此处
  115.     // ㈣ 因为某线程可能重入，需要将 state 全部释放，获取state，然后把它全部减掉，以全部释放
  116.     final int fullyRelease(Node node) {
  117.         boolean failed = true;
  118.         try {
  119.             int savedState = getState();
  120.             // 唤醒等待队列队列中的下一个节点
  121.             if (release(savedState)) {
  122.                 failed = false;
  123.                 return savedState;
  124.             } else {
  125.                 throw new IllegalMonitorStateException();
  126.             }
  127.         } finally {
  128.             if (failed)
  129.                 node.waitStatus = Node.CANCELLED;
  130.         }
  131.     }
  132.     // 打断模式 - 在退出等待时重新设置打断状态
  133.     private static final int REINTERRUPT = 1;
  134.     // 打断模式 - 在退出等待时抛出异常
  135.     private static final int THROW_IE = -1;
  136.     // 判断打断模式
  137.     private int checkInterruptWhileWaiting(Node node) {
  138.         return Thread.interrupted() ?
  139.                 (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :
  140.                 0;
  141.     }
  142.     // ㈤ 应用打断模式
  143.     private void reportInterruptAfterWait(int interruptMode)
  144.             throws InterruptedException {
  145.         if (interruptMode == THROW_IE)
  146.             throw new InterruptedException();
  147.         else if (interruptMode == REINTERRUPT)
  148.             selfInterrupt();
  149.     }
  150.     // 等待 - 直到被唤醒或打断
  151.     public final void await() throws InterruptedException {
  152.         if (Thread.interrupted()) {
  153.             throw new InterruptedException();
  154.         }
  155.         // 添加一个 Node 至等待队列, 见 ㈠
  156.         Node node = addConditionWaiter();
  157.         // 释放节点持有的锁
  158.         int savedState = fullyRelease(node);
  159.         int interruptMode = 0;
  160.         // 如果该节点还没有转移至 AQS 队列, 阻塞
  161.         while (!isOnSyncQueue(node)) {
  162.             // park 阻塞              
  163.             LockSupport.park(this);
  164.             // 如果被打断, 退出等待队列
  165.             if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
  166.                 break;
  167.         }
  168.         // 退出等待队列后, 还需要获得 AQS 队列的锁
  169.         if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
  170.             interruptMode = REINTERRUPT;
  171.         // 所有已取消的 Node 从队列链表删除, 见 ㈡
  172.         if (node.nextWaiter != null)
  173.             unlinkCancelledWaiters();
  174.         // 应用打断模式, 见 ㈤
  175.         if (interruptMode != 0)
  176.             reportInterruptAfterWait(interruptMode);
  177.     }
  178.     // 等待 - 直到被唤醒或打断或超时
  179.     public final long awaitNanos(long nanosTimeout) throws InterruptedException {
  180.         if (Thread.interrupted()) {
  181.             throw new InterruptedException();
  182.         }
  183.         // 添加一个 Node 至等待队列, 见 ㈠
  184.         Node node = addConditionWaiter();
  185.         // 释放节点持有的锁
  186.         int savedState = fullyRelease(node);
  187.         // 获得最后期限
  188.         final long deadline = System.nanoTime() + nanosTimeout;
  189.         int interruptMode = 0;
  190.         // 如果该节点还没有转移至 AQS 队列, 阻塞
  191.         while (!isOnSyncQueue(node)) {
  192.             // 已超时, 退出等待队列
  193.             if (nanosTimeout <= 0L) {
  194.                 transferAfterCancelledWait(node);
  195.                 break;
  196.             }
  197.             // park 阻塞一定时间, spinForTimeoutThreshold 为 1000 ns
  198.             if (nanosTimeout >= spinForTimeoutThreshold)
  199.                 LockSupport.parkNanos(this, nanosTimeout);
  200.             // 如果被打断, 退出等待队列
  201.             if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
  202.                 break;
  203.             nanosTimeout = deadline - System.nanoTime();
  204.         }
  205.         // 退出等待队列后, 还需要获得 AQS 队列的锁
  206.         if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
  207.             interruptMode = REINTERRUPT;
  208.         // 所有已取消的 Node 从队列链表删除, 见 ㈡
  209.         if (node.nextWaiter != null)
  210.             unlinkCancelledWaiters();
  211.         // 应用打断模式, 见 ㈤
  212.         if (interruptMode != 0)
  213.             reportInterruptAfterWait(interruptMode);
  214.         return deadline - System.nanoTime();
  215.     }
  216.     // 等待 - 直到被唤醒或打断或超时, 逻辑类似于 awaitNanos
  217.     public final boolean awaitUntil(Date deadline) throws InterruptedException {
  218.         // ...
  219.     }
  220.     // 等待 - 直到被唤醒或打断或超时, 逻辑类似于 awaitNanos
  221.     public final boolean await(long time, TimeUnit unit) throws InterruptedException {
  222.         // ...
  223.     }
  224.     // 工具方法 省略 ...
  225. }

2.3 读写锁

2.3.1 ReentrantReadWriteLock

当读操作远远高于写操作时，这时候使用读写锁让读-读可以并发，提高性能。读-写，写-写都是相互互斥的！
提供一个数据容器类内部分别使用读锁保护数据的read()方法，写锁保护数据的write()方法 。

读-读测试：

  1. @Slf4j(topic = "c.Test3")
  2. public class Test3 {
  3.     public static void main(String[] args) {
  4.         DataContainer d = new DataContainer();
  5.         new Thread(()->{
  6.             d.read();
  7.         }, "t1").start();
  9.         new Thread(()->{
  10.             d.read();
  11.         }, "t2").start();
  12.     }
  13. }
  16. // 创建读写锁
  17. @Slf4j(topic = "c.DataContainer")
  18. class DataContainer {
  19.     private Object data;
  20.     private ReentrantReadWriteLock rw = new ReentrantReadWriteLock();
  21.     private ReentrantReadWriteLock.ReadLock r = rw.readLock();
  22.     private ReentrantReadWriteLock.WriteLock w = rw.writeLock();
  24.     public Object read(){
  25.         log.debug("获取读锁");
  26.         r.lock();
  27.         try {
  28.             log.debug("读取");
  29.             Thread.sleep(1000);
  31.         } catch (InterruptedException e) {
  32.             e.printStackTrace();
  33.         } finally {
  34.             log.debug("释放读锁");
  35.             r.unlock();
  36.         }
  37.         return data;
  38.     }
  40.     public void write(){
  41.         log.debug("获取写锁");
  42.         w.lock();
  43.         try {
  44.             log.debug("写入");
  45.         }finally {
  46.             log.debug("释放写锁");
  47.             w.unlock();
  48.         }
  50.     }
  1. 17:48:25.705 [t2] DEBUG c.DataContainer - 获取读锁
  2. 17:48:25.705 [t1] DEBUG c.DataContainer - 获取读锁
  3. 17:48:25.708 [t2] DEBUG c.DataContainer - 读取
  4. 17:48:25.708 [t1] DEBUG c.DataContainer - 读取
  5. 17:48:26.712 [t1] DEBUG c.DataContainer - 释放读锁
  6. 17:48:26.712 [t2] DEBUG c.DataContainer - 释放读锁

读是并发的

读写实例

  1. public static void main(String[] args) throws InterruptedException {
  2.         DataContainer d = new DataContainer();
  3.         new Thread(() -> {
  4.             d.read();
  5.         }, "t1").start();
  7.         Thread.sleep(100);
  8.         new Thread(() -> {
  9.             d.write();
  10.         }, "t2").start();
  11.     }
  1. 17:50:56.087 [t1] DEBUG c.DataContainer - 获取读锁
  2. 17:50:56.090 [t1] DEBUG c.DataContainer - 读取
  3. 17:50:56.191 [t2] DEBUG c.DataContainer - 获取写锁
  4. 17:50:57.095 [t1] DEBUG c.DataContainer - 释放读锁
  5. 17:50:57.095 [t2] DEBUG c.DataContainer - 写入
  6. 17:50:57.095 [t2] DEBUG c.DataContainer - 释放写锁

可以看到读与写是互斥的，只有读锁先释放，才能拿到写锁

写-写

  1. @Slf4j(topic = "c.Test3")
  2. public class Test3 {
  3.     public static void main(String[] args) throws InterruptedException {
  4.         DataContainer d = new DataContainer();
  5.         new Thread(() -> {
  6.             d.write();
  7.         }, "t1").start();
  9.         Thread.sleep(100);
  10.         new Thread(() -> {
  11.             d.write();
  12.         }, "t2").start();
  13.     }
  14. }
  17. // 创建读写锁
  18. @Slf4j(topic = "c.DataContainer")
  19. class DataContainer {
  20.     private Object data;
  21.     private ReentrantReadWriteLock rw = new ReentrantReadWriteLock();
  22.     private ReentrantReadWriteLock.ReadLock r = rw.readLock();
  23.     private ReentrantReadWriteLock.WriteLock w = rw.writeLock();
  25.     public Object read() {
  26.         log.debug("获取读锁");
  27.         r.lock();
  28.         try {
  29.             log.debug("读取");
  30.             Thread.sleep(1000);
  32.         } catch (InterruptedException e) {
  33.             e.printStackTrace();
  34.         } finally {
  35.             log.debug("释放读锁");
  36.             r.unlock();
  37.         }
  38.         return data;
  39.     }
  41.     public void write() {
  42.         log.debug("获取写锁");
  43.         w.lock();
  44.         try {
  45.             log.debug("写入");
  46.             Thread.sleep(1000);
  47.         } catch (InterruptedException e) {
  48.             e.printStackTrace();
  49.         } finally {
  50.             log.debug("释放写锁");
  51.             w.unlock();
  52.         }
  54.     }
  55. }
  1. 18:06:20.475 [t1] DEBUG c.DataContainer - 获取写锁
  2. 18:06:20.480 [t1] DEBUG c.DataContainer - 写入
  3. 18:06:20.575 [t2] DEBUG c.DataContainer - 获取写锁
  4. 18:06:21.485 [t1] DEBUG c.DataContainer - 释放写锁
  5. 18:06:21.486 [t2] DEBUG c.DataContainer - 写入
  6. 18:06:22.490 [t2] DEBUG c.DataContainer - 释放写锁

写与写之间也是互斥的。

2.3.2 读写锁注意事项

  1. 读锁不支持条件变量
  2. 重入时升级不支持：即持有读锁的情况下去获取写锁，会导致获取写锁永久等待 ```java r.lock(); try {
    1.   // ...
    2.   w.lock();
    3.   try {
    4.       // ...
    5.   } finally{
    6.       w.unlock();
    7.   }
    } finally{
    1.   r.unlock();
    }
  2. 3. 重入时降级支持：即持有写锁的情况下去获取读锁
  3. ```java
  4. class CachedData {
  5.     Object data;
  6.     // 是否有效，如果失效，需要重新计算 data
  7.     volatile boolean cacheValid;
  8.     final ReentrantReadWriteLock rwl = new ReentrantReadWriteLock();
  9.     void processCachedData() {
  10.         rwl.readLock().lock();
  11.         if (!cacheValid) {
  12.             // 获取写锁前必须释放读锁
  13.             rwl.readLock().unlock();
  14.             rwl.writeLock().lock();
  15.             try {
  16.                 // 判断是否有其它线程已经获取了写锁、更新了缓存, 避免重复更新
  17.                 if (!cacheValid) {
  18.                     data = ...
  19.                     cacheValid = true;
  20.                 }
  21.                 // 降级为读锁, 释放写锁, 这样能够让其它线程读取缓存
  22.                 rwl.readLock().lock();
  23.             } finally {
  25.                 rwl.writeLock().unlock();
  26.             }
  27.         }
  28.         // 自己用完数据, 释放读锁
  29.         try {
  30.             use(data);
  31.         } finally {
  32.             rwl.readLock().unlock();
  33.         }
  34.     }
  35. }

2.3.3 应用之缓存

缓存更新策略：更新时，是先清缓存还是先更新数据库？
先清除缓存操作如下：
image.png
先更新数据库操作如下：
image.png
补充一种情况，假设查询线程 A 查询数据时恰好缓存数据由于时间到期失效，或是第一次查询：这种情况的出现几率非常小：
image.png

  1. /**
  2.  * ReentrantReadWriteLock 读写锁解决 缓存与数据库一致性问题
  3.  */
  4. public class Code_13_ReadWriteCacheTest {
  5.     public static void main(String[] args) {
  7.         GeneriCacheDao<Object> generiCacheDao = new GeneriCacheDao<>();
  9.         Object[] objects = new Object[2];
  10.         generiCacheDao.queryOne(Object.class,"Test",objects);
  11.         generiCacheDao.queryOne(Object.class,"Test",objects);
  12.         generiCacheDao.queryOne(Object.class,"Test",objects);
  13.         generiCacheDao.queryOne(Object.class,"Test",objects);
  14.         System.out.println(generiCacheDao.map);
  15.         generiCacheDao.update("Test",objects);
  16.         System.out.println(generiCacheDao.map);
  17.     }
  19. }
  21. class GeneriCacheDao<T>  extends GenericDao {
  23.     HashMap<SqlPair, T> map = new HashMap<>();
  24.     ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
  25.     GenericDao genericDao = new GenericDao();
  27.     @Override
  28.     public int update(String sql, Object... params){
  29.         lock.writeLock().lock();
  30.         SqlPair sqlPair = new SqlPair(sql, params);
  31.         try {
  32.             // 先查询数据库再更新缓存,但是这里加了锁，谁先谁后都没关系
  33.             int update = genericDao.update(sql, params);
  34.             map.clear();
  35.             return update;
  36.         } finally {
  37.             lock.writeLock().unlock();
  38.         }
  39.     }
  41.     @Override
  42.     public T queryOne(Class beanClass, String sql, Object... params){
  43.         SqlPair key = new SqlPair(sql, params);
  44.         // 加读锁, 防止其它线程对缓存更改
  45.         lock.readLock().lock();
  47.         try {
  48.             T t = map.get(key);
  49.             if (t != null){
  50.                 return t;
  51.             }
  52.         } finally {
  53.             lock.readLock().unlock();
  54.         }
  56.         // 加写锁, 防止其它线程对缓存读取和更改
  57.         lock.writeLock().lock();
  58.         // get 方法上面部分是可能多个线程进来的, 可能已经向缓存填充了数据
  59.         // 为防止重复查询数据库, 再次验证
  60.         try {
  61.             T value = map.get(key);
  62.             if (value == null){
  63.                 value = (T) genericDao.queryOne(beanClass, sql, params);
  64.                 map.put(key, value);
  65.             }
  66.             return value;
  67.         } finally {
  68.             lock.writeLock().unlock();
  69.         }
  70.     }
  73.     class SqlPair{
  74.         private String sql;
  76.         private Object[] params;
  78.         public SqlPair(String sql, Object[] params) {
  79.             this.sql = sql;
  80.             this.params = params;
  81.         }
  83.         @Override
  84.         public boolean equals(Object o) {
  85.             if (this == o) return true;
  86.             if (o == null || getClass() != o.getClass()) return false;
  87.             SqlPair sqlMap = (SqlPair) o;
  88.             return Objects.equals(sql, sqlMap.sql) &&
  89.                     Arrays.equals(params, sqlMap.params);
  90.         }
  92.         @Override
  93.         public int hashCode() {
  94.             int result = Objects.hash(sql);
  95.             result = 31 * result + Arrays.hashCode(params);
  96.             return result;
  97.         }
  98.     }
  99. }
  101. class GenericDao<T>{
  102.     public int update(String sql, Object... params){
  103.         return 1;
  104.     }
  106.     public T queryOne(Class<T> beanClass, String sql, Object... params){
  107.         System.out.println("查询数据库中");
  108.         return (T) new Object();
  109.     }
  110. }

2.3.4 读写锁原理

图解流程

读写锁用的是同一个 Sync 同步器，因此等待队列、state 等也是同一个
下面执行：t1 w.lock，t2 r.lock 情况

  1. t1 成功上锁，流程与 ReentrantLock 加锁相比没有特殊之处，不同是写锁状态占了 state 的低 16 位，而读锁使用的是 state 的高 16 位

image.png

  1. t2 执行 r.lock，这时进入读锁的 sync.acquireShared(1) 流程，首先会进入 tryAcquireShared 流程。如果有写锁占据，那么 tryAcquireShared 返回 -1 表示失败。tryAcquireShared 返回值表示
    • -1 表示失败
    • 0 表示成功，但后继节点不会继续唤醒
    • 正数表示成功，而且数值是还有几个后继节点需要唤醒，我们这里的读写锁返回 1

image.png

  1. 这时会进入 sync.doAcquireShared(1) 流程，首先也是调用 addWaiter 添加节点，不同之处在于节点被设置为 Node.SHARED 模式而非 Node.EXCLUSIVE 模式，注意此时 t2 仍处于活跃状态

image.png

  1. t2 会看看自己的节点是不是老二，如果是，还会再次调用 tryAcquireShared(1) 来尝试获取锁
  2. 如果没有成功，在 doAcquireShared 内 for (;;) 循环一次，把前驱节点的 waitStatus 改为 -1，再 for (;;) 循环一次尝试 tryAcquireShared(1) 如果还不成功，那么在 parkAndCheckInterrupt() 处 park。

image.png
又继续执行：t3 r.lock，t4 w.lock。这种状态下，假设又有 t3 加读锁和 t4 加写锁，这期间 t1 仍然持有锁，就变成了下面的样子
image.png
继续执行 t1 w.unlock
这时会走到写锁的 sync.release(1) 流程，调用 sync.tryRelease(1) 成功，变成下面的样子
image.png
接下来执行唤醒流程 sync.unparkSuccessor，即让老二恢复运行，这时 t2 在 doAcquireShared 内 parkAndCheckInterrupt() 处恢复运行，图中的t2从黑色变成了蓝色（注意这里只是恢复运行而已，并没有获取到锁！） 这回再来一次 for (;; ) 执行 tryAcquireShared 成功则让读锁计数加
image.png
这时 t2 已经恢复运行，接下来 t2 调用 setHeadAndPropagate(node, 1)，它原本所在节点被置为头节点
image.png
事情还没完，在 setHeadAndPropagate 方法内还会检查下一个节点是否是 shared，如果是则调用 doReleaseShared() 将 head 的状态从 -1 改为 0 并唤醒老二，这时 t3 在 doAcquireShared 内 parkAndCheckInterrupt() 处恢复运行.(读锁是共享的，不是独占的，多个线程都能读)
image.png
这回再来一次 for (;; ) 执行 tryAcquireShared 成功则让读锁计数加一
image.png
这时 t3 已经恢复运行，接下来 t3 调用 setHeadAndPropagate(node, 1)，它原本所在节点被置为头节点
image.png
再继续执行t2 r.unlock，t3 r.unlock 。t2进入 sync.releaseShared(1) 中，调用 tryReleaseShared(1) 让计数减一，但由于计数还不为零
image.png
t3 进入 sync.releaseShared(1) 中，调用 tryReleaseShared(1) 让计数减一，这回计数为零了，进入 doReleaseShared() 将头节点从 -1 改为 0 并唤醒老二，即
image.png
之后 t4 在 acquireQueued 中 parkAndCheckInterrupt 处恢复运行，再次 for (;; ) 这次自己是老二，并且没有其他 竞争，tryAcquire(1) 成功，修改头结点，流程结束
image.png

源码分析

  • 写锁上锁流程 ```java static final class NonfairSync extends Sync { // … 省略无关代码

    // 外部类 WriteLock 方法, 方便阅读, 放在此处 public void lock() {

    1.   sync.acquire(1);

    }

    // AQS 继承过来的方法, 方便阅读, 放在此处 public final void acquire(int arg) {

    1.   if (
    2.       // 尝试获得写锁失败
    3.           !tryAcquire(arg) &&
    4.                   // 将当前线程关联到一个 Node 对象上, 模式为独占模式
    5.                   // 进入 AQS 队列阻塞
    6.                   acquireQueued(addWaiter(Node.EXCLUSIVE), arg)
    7.   ) {
    8.       selfInterrupt();
    9.   }

    }

    // Sync 继承过来的方法, 方便阅读, 放在此处 protected final boolean tryAcquire(int acquires) {

    1.   // 获得低 16 位, 代表写锁的 state 计数
    2.   Thread current = Thread.currentThread();
    3.   int c = getState();
    4.   int w = exclusiveCount(c);
    6.   if (c != 0) {
    7.       if (
    8.           // c != 0 and w == 0 表示有读锁返回错误，读锁不支持锁升级, 或者
    9.               w == 0 ||
    10.                       // c != 0 and w != 0 表示有写锁，如果 exclusiveOwnerThread 不是自己
    11.                       current != getExclusiveOwnerThread()
    12.       ) {
    13.           // 获得锁失败
    14.           return false;
    15.       }
    16.       // 写锁计数超过低 16 位, 报异常
    17.       if (w + exclusiveCount(acquires) > MAX_COUNT)
    18.           throw new Error("Maximum lock count exceeded");
    19.       // 写锁重入, 获得锁成功
    20.       setState(c + acquires);
    21.       return true;
    22.   }
    23.   if (
    24.       // 判断写锁是否该阻塞这里返回false, 或者
    25.           writerShouldBlock() ||
    26.                   // 尝试更改计数失败
    27.                   !compareAndSetState(c, c + acquires)
    28.   ) {
    29.       // 获得锁失败
    30.       return false;
    31.   }
    32.   // 获得锁成功
    33.   setExclusiveOwnerThread(current);
    34.   return true;

    }

    // 非公平锁 writerShouldBlock 总是返回 false, 无需阻塞 final boolean writerShouldBlock() {

    1.   return false;

    } }

  2. - 写锁释放过程
  3. ```java
  4. static final class NonfairSync extends Sync {
  5.     // ... 省略无关代码
  7.     // WriteLock 方法, 方便阅读, 放在此处
  8.     public void unlock() {
  9.         sync.release(1);
  10.     }
  12.     // AQS 继承过来的方法, 方便阅读, 放在此处
  13.     public final boolean release(int arg) {
  14.         // 尝试释放写锁成功
  15.         if (tryRelease(arg)) {
  16.             // unpark AQS 中等待的线程
  17.             Node h = head;
  18.             if (h != null && h.waitStatus != 0)
  19.                 unparkSuccessor(h);
  20.             return true;
  21.         }
  22.         return false;
  23.     }
  25.     // Sync 继承过来的方法, 方便阅读, 放在此处
  26.     protected final boolean tryRelease(int releases) {
  27.         if (!isHeldExclusively())
  28.             throw new IllegalMonitorStateException();
  29.         int nextc = getState() - releases;
  30.         // 因为可重入的原因, 写锁计数为 0, 才算释放成功
  31.         boolean free = exclusiveCount(nextc) == 0;
  32.         if (free) {
  33.             setExclusiveOwnerThread(null);
  34.         }
  35.         setState(nextc);
  36.         return free;
  37.     }
  38. }

  • 读锁上锁过程 ```java static final class NonfairSync extends Sync {

    // ReadLock 方法, 方便阅读, 放在此处 public void lock() {

    1.   sync.acquireShared(1);

    }

    // AQS 继承过来的方法, 方便阅读, 放在此处 public final void acquireShared(int arg) {

    1.   // tryAcquireShared 返回负数, 表示获取读锁失败
    2.   if (tryAcquireShared(arg) < 0) {
    3.       doAcquireShared(arg);
    4.   }

    }

    // Sync 继承过来的方法, 方便阅读, 放在此处 protected final int tryAcquireShared(int unused) {

    1.   Thread current = Thread.currentThread();
    2.   int c = getState();
    3.   // 如果是其它线程持有写锁, 获取读锁失败
    4.   if (
    5.           exclusiveCount(c) != 0 &&
    6.                   getExclusiveOwnerThread() != current
    7.   ) {
    8.       return -1;
    9.   }
    10.   int r = sharedCount(c);
    11.   if (
    12.       // 读锁不该阻塞(如果老二是写锁，读锁该阻塞), 并且
    13.           !readerShouldBlock() &&
    14.                   // 小于读锁计数, 并且
    15.                   r < MAX_COUNT &&
    16.                   // 尝试增加计数成功
    17.                   compareAndSetState(c, c + SHARED_UNIT)
    18.   ) {
    19.       // ... 省略不重要的代码
    20.       return 1;
    21.   }
    22.   return fullTryAcquireShared(current);

    }

    // 非公平锁 readerShouldBlock 看 AQS 队列中第一个节点是否是写锁 // true 则该阻塞, false 则不阻塞 final boolean readerShouldBlock() {

    1.   return apparentlyFirstQueuedIsExclusive();

    }

    // AQS 继承过来的方法, 方便阅读, 放在此处 // 与 tryAcquireShared 功能类似, 但会不断尝试 for (;;) 获取读锁, 执行过程中无阻塞 final int fullTryAcquireShared(Thread current) {

    1.   HoldCounter rh = null;
    2.   for (;;) {
    3.       int c = getState();
    4.       if (exclusiveCount(c) != 0) {
    5.           if (getExclusiveOwnerThread() != current)
    6.               return -1;
    7.       } else if (readerShouldBlock()) {
    8.           // ... 省略不重要的代码
    9.       }
    10.       if (sharedCount(c) == MAX_COUNT)
    11.           throw new Error("Maximum lock count exceeded");
    12.       if (compareAndSetState(c, c + SHARED_UNIT)) {
    13.           // ... 省略不重要的代码
    14.           return 1;
    15.       }
    16.   }

    }

    // AQS 继承过来的方法, 方便阅读, 放在此处 private void doAcquireShared(int arg) {

    1.   // 将当前线程关联到一个 Node 对象上, 模式为共享模式
    2.   final Node node = addWaiter(Node.SHARED);
    3.   boolean failed = true;
    4.   try {
    5.       boolean interrupted = false;
    6.       for (;;) {
    7.           final Node p = node.predecessor();
    8.           if (p == head) {
    9.               // 再一次尝试获取读锁
    10.               int r = tryAcquireShared(arg);
    11.               // 成功
    12.               if (r >= 0) {
    13.                   // ㈠
    14.                   // r 表示可用资源数, 在这里总是 1 允许传播
    15.                   //（唤醒 AQS 中下一个 Share 节点）
    16.                   setHeadAndPropagate(node, r);
    17.                   p.next = null; // help GC
    18.                   if (interrupted)
    19.                       selfInterrupt();
    20.                   failed = false;
    21.                   return;
    22.               }
    23.           }
    24.           if (
    25.               // 是否在获取读锁失败时阻塞（前一个阶段 waitStatus == Node.SIGNAL）
    26.                   shouldParkAfterFailedAcquire(p, node) &&
    27.                           // park 当前线程
    28.                           parkAndCheckInterrupt()
    29.           ) {
    30.               interrupted = true;
    31.           }
    32.       }
    33.   } finally {
    34.       if (failed)
    35.           cancelAcquire(node);
    36.   }

    }

    // ㈠ AQS 继承过来的方法, 方便阅读, 放在此处 private void setHeadAndPropagate(Node node, int propagate) {

    1.   Node h = head; // Record old head for check below
    2.   // 设置自己为 head
    3.   setHead(node);
    5.   // propagate 表示有共享资源（例如共享读锁或信号量）
    6.   // 原 head waitStatus == Node.SIGNAL 或 Node.PROPAGATE
    7.   // 现在 head waitStatus == Node.SIGNAL 或 Node.PROPAGATE
    8.   if (propagate > 0 || h == null || h.waitStatus < 0 ||
    9.           (h = head) == null || h.waitStatus < 0) {
    10.       Node s = node.next;
    11.       // 如果是最后一个节点或者是等待共享读锁的节点
    12.       if (s == null || s.isShared()) {
    13.           // 进入 ㈡
    14.           doReleaseShared();
    15.       }
    16.   }

    }

    // ㈡ AQS 继承过来的方法, 方便阅读, 放在此处 private void doReleaseShared() {

    1.   // 如果 head.waitStatus == Node.SIGNAL ==> 0 成功, 下一个节点 unpark
    2.   // 如果 head.waitStatus == 0 ==> Node.PROPAGATE, 为了解决 bug, 见后面分析，参考这里：http://www.tianxiaobo.com/2018/05/01/AbstractQueuedSynchronizer-%E5%8E%9F%E7%90%86%E5%88%86%E6%9E%90-%E7%8B%AC%E5%8D%A0-%E5%85%B1%E4%BA%AB%E6%A8%A1%E5%BC%8F/#5propagate-%E7%8A%B6%E6%80%81%E5%AD%98%E5%9C%A8%E7%9A%84%E6%84%8F%E4%B9%89
    3.   for (;;) {
    4.       Node h = head;
    5.       // 队列还有节点
    6.       if (h != null && h != tail) {
    7.           int ws = h.waitStatus;
    8.           if (ws == Node.SIGNAL) {
    9.               if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
    10.                   continue; // loop to recheck cases
    11.               // 下一个节点 unpark 如果成功获取读锁
    12.               // 并且下下个节点还是 shared, 继续 doReleaseShared
    13.               unparkSuccessor(h);
    14.           }
    15.           else if (ws == 0 &&
    16.                   !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
    17.               continue; // loop on failed CAS
    18.       }
    19.       if (h == head) // loop if head changed
    20.           break;
    21.   }

    } }

  2. - 读锁释放过程：
  3. ```java
  4. static final class NonfairSync extends Sync {
  6.     // ReadLock 方法, 方便阅读, 放在此处
  7.     public void unlock() {
  8.         sync.releaseShared(1);
  9.     }
  11.     // AQS 继承过来的方法, 方便阅读, 放在此处
  12.     public final boolean releaseShared(int arg) {
  13.         if (tryReleaseShared(arg)) {
  14.             doReleaseShared();
  15.             return true;
  16.         }
  17.         return false;
  18.     }
  20.     // Sync 继承过来的方法, 方便阅读, 放在此处
  21.     protected final boolean tryReleaseShared(int unused) {
  22.         // ... 省略不重要的代码
  23.         for (;;) {
  24.             int c = getState();
  25.             int nextc = c - SHARED_UNIT;
  26.             if (compareAndSetState(c, nextc)) {
  27.                 // 读锁的计数不会影响其它获取读锁线程, 但会影响其它获取写锁线程
  28.                 // 计数为 0 才是真正释放
  29.                 return nextc == 0;
  30.             }
  31.         }
  32.     }
  34.     // AQS 继承过来的方法, 方便阅读, 放在此处
  35.     private void doReleaseShared() {
  36.         // 如果 head.waitStatus == Node.SIGNAL ==> 0 成功, 下一个节点 unpark
  37.         // 如果 head.waitStatus == 0 ==> Node.PROPAGATE
  38.         for (;;) {
  39.             Node h = head;
  40.             if (h != null && h != tail) {
  41.                 int ws = h.waitStatus;
  42.                 // 如果有其它线程也在释放读锁，那么需要将 waitStatus 先改为 0
  43.                 // 防止 unparkSuccessor 被多次执行
  44.                 if (ws == Node.SIGNAL) {
  45.                     if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0))
  46.                         continue; // loop to recheck cases
  47.                     unparkSuccessor(h);
  48.                 }
  49.                 // 如果已经是 0 了，改为 -3，用来解决传播性，见后文信号量 bug 分析
  50.                 else if (ws == 0 &&
  51.                         !compareAndSetWaitStatus(h, 0, Node.PROPAGATE))
  52.                     continue; // loop on failed CAS
  53.             }
  54.             if (h == head) // loop if head changed
  55.                 break;
  56.         }
  57.     }
  58. }

2.3.5 StampedLock

该类自 JDK 8 加入，是为了进一步优化读性能，它的特点是在使用读锁、写锁时都必须配合【戳】使用。StampedLock 不支持条件变量；StampedLock 不支持可重入

  • 加解读锁 ```java long stamp = lock.readLock(); lock.unlockRead(stamp);
  2. - 加解写锁
  3. ```java
  4. long stamp = lock.writeLock();
  5. lock.unlockWrite(stamp);
  • 乐观读，StampedLock 支持 tryOptimisticRead() 方法（乐观读），读取完毕后需要做一次 戳校验 如果校验通 过，表示这期间确实没有写操作，数据可以安全使用，如果校验没通过，需要重新获取读锁，保证数据安全。 ```java long stamp = lock.tryOptimisticRead(); // 验戳 if(!lock.validate(stamp)){ // 锁升级 }
  2. - 代码示例：
  4. 提供一个 数据容器类 内部分别使用读锁保护数据的 read() 方法，写锁保护数据的 write() 方法。
  5. ```java
  6. public class Code_14_StampedLockTest {
  8.     public static void main(String[] args) throws InterruptedException {
  9.         StampedLockDataContainer dataContainer = new StampedLockDataContainer(1);
  11.         Thread t1 = new Thread(() -> {
  12.             try {
  13.                 System.out.println(dataContainer.read(1));
  14.             } catch (InterruptedException e) {
  15.                 e.printStackTrace();
  16.             }
  17.         }, "t1");
  19.         t1.start();
  21.         TimeUnit.MILLISECONDS.sleep(500);
  23.         Thread t2 = new Thread(() -> {
  24. //            try {
  25. //                dataContainer.read(0);
  26. //            } catch (InterruptedException e) {
  27. //                e.printStackTrace();
  28. //            }
  29.             dataContainer.write(10);
  30.         }, "t2");
  32.         t2.start();
  33.     }
  35. }
  37. @Slf4j(topic = "c.StampedLockDataContainer")
  38. class StampedLockDataContainer {
  40.     private int data;
  41.     private StampedLock stampedLock = new StampedLock();
  43.     public StampedLockDataContainer(int data) {
  44.         this.data = data;
  45.     }
  47.     public int read(int readTime) throws InterruptedException {
  48.         long stamp = stampedLock.tryOptimisticRead();
  49.         log.info("optimistic read locking ...{}", stamp);
  50.         Thread.sleep(readTime * 1000);
  51.         if(stampedLock.validate(stamp)) {
  52.             log.info("read finish... {}", stamp);
  53.             return data;
  54.         }
  55.         // 锁升级 - 读锁
  56.         log.info("update to read lock ...");
  57.         try {
  58.             stamp = stampedLock.readLock();
  59.             log.info("read lock {}", stamp);
  60.             Thread.sleep(readTime * 1000);
  61.             log.info("read finish ... {}", stamp);
  62.             return data;
  63.         } finally {
  64.             stampedLock.unlockRead(stamp);
  65.         }
  66.     }
  68.     public void write(int newData) {
  69.         long stamp = stampedLock.writeLock();
  70.         try {
  71.             log.info("write lock {}", stamp);
  72.             this.data = newData;
  73.             try {
  74.                 TimeUnit.SECONDS.sleep(1);
  75.             } catch (InterruptedException e) {
  76.                 e.printStackTrace();
  77.             }
  78.             log.info("write finish ... {}", stamp);
  79.             log.info("write newData ... {}", this.data);
  80.         } finally {
  81.             stampedLock.unlockWrite(stamp);
  82.         }
  83.     }
  85. }

2.4 Semaphore信号量

2.4.1 基本使用

信号量，用来限制能同时访问共享资源的线程上限。

  1. @Slf4j(topic = "c.Test4")
  2. public class Test4 {
  3.     public static void main(String[] args) {
  4.         // 创建 semaphore对象
  5.         // 同时运行的上限是3个
  6.         Semaphore semaphore = new Semaphore(3);
  8.         // 创建10个线程同时运行
  9.         for (int i = 0; i < 10; i++) {
  10.             new Thread(()->{
  11.                 // 获取许可
  12.                 try {
  13.                     semaphore.acquire();
  14.                 } catch (InterruptedException e) {
  15.                     e.printStackTrace();
  16.                 }
  17.                 // 运行
  18.                 log.debug("running");
  19.                 try {
  20.                     Thread.sleep(1000);
  21.                     log.debug("end...");
  22.                 } catch (InterruptedException e) {
  23.                     e.printStackTrace();
  24.                 }finally {
  25.                     // 释放许可
  26.                     semaphore.release();
  27.                 }
  29.             }).start();
  30.         }
  31.     }
  32. }
  1. 16:56:59.721 [Thread-0] DEBUG c.Test4 - running
  2. 16:56:59.721 [Thread-2] DEBUG c.Test4 - running
  3. 16:56:59.721 [Thread-1] DEBUG c.Test4 - running
  4. 16:57:00.730 [Thread-1] DEBUG c.Test4 - end...
  5. 16:57:00.731 [Thread-2] DEBUG c.Test4 - end...
  6. 16:57:00.731 [Thread-0] DEBUG c.Test4 - end...
  7. 16:57:00.731 [Thread-3] DEBUG c.Test4 - running
  8. 16:57:00.731 [Thread-5] DEBUG c.Test4 - running
  9. 16:57:00.732 [Thread-4] DEBUG c.Test4 - running
  10. 16:57:01.736 [Thread-5] DEBUG c.Test4 - end...
  11. 16:57:01.736 [Thread-4] DEBUG c.Test4 - end...
  12. 16:57:01.736 [Thread-3] DEBUG c.Test4 - end...
  13. 16:57:01.736 [Thread-6] DEBUG c.Test4 - running
  14. 16:57:01.736 [Thread-7] DEBUG c.Test4 - running
  15. 16:57:01.736 [Thread-8] DEBUG c.Test4 - running
  16. 16:57:02.740 [Thread-8] DEBUG c.Test4 - end...
  17. 16:57:02.740 [Thread-7] DEBUG c.Test4 - end...
  18. 16:57:02.740 [Thread-6] DEBUG c.Test4 - end...
  19. 16:57:02.741 [Thread-9] DEBUG c.Test4 - running
  20. 16:57:03.744 [Thread-9] DEBUG c.Test4 - end...

2.4.2 图解流程

Semaphore 有点像一个停车场，permits 就好像停车位数量，当线程获得了 permits 就像是获得了停车位，然后停车场显示空余车位减一刚开始，permits（state）为 3，这时 5 个线程来获取资源。
image.png
假设其中 Thread-1，Thread-2，Thread-4 cas 竞争成功，而 Thread-0 和 Thread-3 竞争失败，进入 AQS 队列park 阻塞
image.png
这时 Thread-4 释放了 permits，状态如下
image.png
接下来 Thread-0 竞争成功，permits 再次设置为 0，设置自己为 head 节点，断开原来的 head 节点，unpark 接下来的 Thread-3 节点，但由于 permits 是 0，因此 Thread-3 在尝试不成功后再次进入 park 状态
image.png

2.4.3 源码

  1. static final class NonfairSync extends Sync {
  2.     private static final long serialVersionUID = -2694183684443567898L;
  3.     NonfairSync(int permits) {
  4.         // permits 即 state
  5.         super(permits);
  6.     }
  8.     // Semaphore 方法, 方便阅读, 放在此处
  9.     public void acquire() throws InterruptedException {
  10.         sync.acquireSharedInterruptibly(1);
  11.     }
  12.     // AQS 继承过来的方法, 方便阅读, 放在此处
  13.     public final void acquireSharedInterruptibly(int arg)
  14.             throws InterruptedException {
  15.         if (Thread.interrupted())
  16.             throw new InterruptedException();
  17.         if (tryAcquireShared(arg) < 0)
  18.             doAcquireSharedInterruptibly(arg);
  19.     }
  21.     // 尝试获得共享锁
  22.     protected int tryAcquireShared(int acquires) {
  23.         return nonfairTryAcquireShared(acquires);
  24.     }
  26.     // Sync 继承过来的方法, 方便阅读, 放在此处
  27.     final int nonfairTryAcquireShared(int acquires) {
  28.         for (;;) {
  29.             int available = getState();
  30.             int remaining = available - acquires;
  31.             if (
  32.                 // 如果许可已经用完, 返回负数, 表示获取失败, 进入 doAcquireSharedInterruptibly
  33.                     remaining < 0 ||
  34.                             // 如果 cas 重试成功, 返回正数, 表示获取成功
  35.                             compareAndSetState(available, remaining)
  36.             ) {
  37.                 return remaining;
  38.             }
  39.         }
  40.     }
  42.     // AQS 继承过来的方法, 方便阅读, 放在此处
  43.     private void doAcquireSharedInterruptibly(int arg) throws InterruptedException {
  44.         final Node node = addWaiter(Node.SHARED);
  45.         boolean failed = true;
  46.         try {
  47.             for (;;) {
  48.                 final Node p = node.predecessor();
  49.                 if (p == head) {
  50.                     // 再次尝试获取许可
  51.                     int r = tryAcquireShared(arg);
  52.                     if (r >= 0) {
  53.                         // 成功后本线程出队（AQS）, 所在 Node设置为 head
  54.                         // 如果 head.waitStatus == Node.SIGNAL ==> 0 成功, 下一个节点 unpark
  55.                         // 如果 head.waitStatus == 0 ==> Node.PROPAGATE
  56.                       // r 表示可用资源数, 为 0 则不会继续传播
  57.                         setHeadAndPropagate(node, r);
  58.                         p.next = null; // help GC
  59.                         failed = false;
  60.                         return;
  61.                     }
  62.                 }
  63.                 // 不成功, 设置上一个节点 waitStatus = Node.SIGNAL, 下轮进入 park 阻塞
  64.                 if (shouldParkAfterFailedAcquire(p, node) &&
  65.                         parkAndCheckInterrupt())
  66.                     throw new InterruptedException();
  67.             }
  68.         } finally {
  69.             if (failed)
  70.                 cancelAcquire(node);
  71.         }
  72.     }
  74.     // Semaphore 方法, 方便阅读, 放在此处
  75.     public void release() {
  76.         sync.releaseShared(1);
  77.     }
  79.     // AQS 继承过来的方法, 方便阅读, 放在此处
  80.     public final boolean releaseShared(int arg) {
  81.         if (tryReleaseShared(arg)) {
  82.             doReleaseShared();
  83.             return true;
  84.         }
  85.         return false;
  86.     }
  88.     // Sync 继承过来的方法, 方便阅读, 放在此处
  89.     protected final boolean tryReleaseShared(int releases) {
  90.         for (;;) {
  91.             int current = getState();
  92.             int next = current + releases;
  93.             if (next < current) // overflow
  94.                 throw new Error("Maximum permit count exceeded");
  95.             if (compareAndSetState(current, next))
  96.                 return true;
  97.         }
  98.     }
  99. }

2.5 CountdownLatch 倒计时锁

  • CountDownLatch 允许多线程阻塞在一个地方，直至所有线程的任务都执行完毕。在 Java 并发中，countdownlatch 的概念是一个常见的面试题，所以一定要确保你很好的理解了它。
  • CountDownLatch 是共享锁的一种实现,它默认构造 AQS 的 state 值为 count。当线程使用 countDown方法时,其实使用了 tryReleaseShared 方法以CAS 的操作来减少 state ,直至 state 为 0 就代表所有的线程都调用了countDown方法。当调用 await 方法的时候，如果 state 不为0，就代表仍然有线程没有调用 countDown 方法，那么就把已经调用过 countDown 的线程都放入阻塞队列 Park ,并自旋 CAS 判断 state == 0，直至最后一个线程调用了 countDown ，使得 state == 0，于是阻塞的线程便判断成功，全部往下执行。

  • 用来进行线程同步协作，等待所有线程完成倒计时。 其中构造参数用来初始化等待计数值，await() 用来等待计数归零，countDown() 用来让计数减一。

    1. @Slf4j(topic = "c.test5")
    2. public class Test5 {
    4.   public static void main(String[] args) throws InterruptedException {
    5.       CountDownLatch latch = new CountDownLatch(3);
    7.       new Thread(()->{
    8.           log.debug("begin..");
    9.           try {
    10.               Thread.sleep(1000);
    11.               latch.countDown();
    12.               log.debug("end...,{}",latch.getCount());
    13.           } catch (InterruptedException e) {
    14.               e.printStackTrace();
    15.           }
    16.       }).start();
    18.       new Thread(()->{
    19.           log.debug("begin..");
    20.           try {
    21.               Thread.sleep(2000);
    22.               latch.countDown();
    23.               log.debug("end...,{}",latch.getCount());
    24.           } catch (InterruptedException e) {
    25.               e.printStackTrace();
    26.           }
    27.       }).start();
    29.       new Thread(()->{
    30.           log.debug("begin..");
    31.           try {
    32.               Thread.sleep(1500);
    33.               latch.countDown();
    34.               log.debug("end...,{}",latch.getCount());
    35.           } catch (InterruptedException e) {
    36.               e.printStackTrace();
    37.           }
    38.       }).start();
    40.       log.debug("等待...");
    41.       latch.await();
    42.       log.debug("等待结束");
    43.   }
    44. }
    1. 17:31:13.539 [Thread-1] DEBUG c.test5 - begin..
    2. 17:31:13.539 [Thread-2] DEBUG c.test5 - begin..
    3. 17:31:13.539 [main] DEBUG c.test5 - 等待...
    4. 17:31:13.539 [Thread-0] DEBUG c.test5 - begin..
    5. 17:31:14.546 [Thread-0] DEBUG c.test5 - end...,2
    6. 17:31:15.045 [Thread-2] DEBUG c.test5 - end...,1
    7. 17:31:15.547 [Thread-1] DEBUG c.test5 - end...,0
    8. 17:31:15.547 [main] DEBUG c.test5 - 等待结束

    每一个线程会countDown一次，最终到0为止

    使用线程池

  2. @Slf4j(topic = "c.test5")
  3. public class Test5 {
  5.     public static void main(String[] args) throws InterruptedException {
  6.         CountDownLatch latch = new CountDownLatch(3);
  7.         ExecutorService service = Executors.newFixedThreadPool(4);
  9.         service.submit(()->{
  10.             log.debug("begin..");
  11.             try {
  12.                 Thread.sleep(1000);
  13.                 latch.countDown();
  14.                 log.debug("end...,{}",latch.getCount());
  15.             } catch (InterruptedException e) {
  16.                 e.printStackTrace();
  17.             }
  18.         });
  20.         service.submit(()->{
  21.             log.debug("begin..");
  22.             try {
  23.                 Thread.sleep(2000);
  24.                 latch.countDown();
  25.                 log.debug("end...,{}",latch.getCount());
  26.             } catch (InterruptedException e) {
  27.                 e.printStackTrace();
  28.             }
  29.         });
  32.         service.submit(()->{
  33.             log.debug("begin..");
  34.             try {
  35.                 Thread.sleep(1500);
  36.                 latch.countDown();
  37.                 log.debug("end...,{}",latch.getCount());
  38.             } catch (InterruptedException e) {
  39.                 e.printStackTrace();
  40.             }
  41.         });
  43.         service.submit(()->{
  44.             log.debug("等待...");
  45.             try {
  46.                 latch.await();
  47.             } catch (InterruptedException e) {
  48.                 e.printStackTrace();
  49.             }
  50.             log.debug("等待结束");
  51.         });
  52.     }
  53. }
  1. 17:34:34.681 [pool-1-thread-3] DEBUG c.test5 - begin..
  2. 17:34:34.681 [pool-1-thread-1] DEBUG c.test5 - begin..
  3. 17:34:34.681 [pool-1-thread-2] DEBUG c.test5 - begin..
  4. 17:34:34.681 [pool-1-thread-4] DEBUG c.test5 - 等待...
  5. 17:34:35.690 [pool-1-thread-1] DEBUG c.test5 - end...,2
  6. 17:34:36.190 [pool-1-thread-3] DEBUG c.test5 - end...,1
  7. 17:34:36.690 [pool-1-thread-2] DEBUG c.test5 - end...,0
  8. 17:34:36.690 [pool-1-thread-4] DEBUG c.test5 - 等待结束

2.6 CyclicBarrier

循环栅栏，用来进行线程协作，等待线程满足某个计数。构造时设置『计数个数』，每个线程执行到某个需要“同步”的时刻调用 await() 方法进行等待，当等待的线程数满足『计数个数』时，继续执行。跟 CountdownLatch 一样，但这个可以重用。

  1. public static void main(String[] args) {
  3.         ExecutorService executorService = Executors.newFixedThreadPool(2);
  4.         CyclicBarrier cyclicBarrier = new CyclicBarrier(2, () -> {
  5.             log.info("task2 finish ...");
  6.         });
  8.         for(int i = 0; i < 3; i++) {
  9.             executorService.submit(() -> {
  10.                 log.info("task1 begin ...");
  11.                 try {
  12.                     Thread.sleep(1000);
  13.                     cyclicBarrier.await();
  14.                 } catch (InterruptedException | BrokenBarrierException e) {
  15.                     e.printStackTrace();
  16.                 }
  17.             });
  18.             executorService.submit(() -> {
  19.                 log.info("task2 begin ...");
  20.                 try {
  21.                     Thread.sleep(2000);
  22.                     cyclicBarrier.await();
  23.                 } catch (InterruptedException | BrokenBarrierException e) {
  24.                     e.printStackTrace();
  25.                 }
  26.             });
  27.         }
  29.         executorService.shutdown();
  30.     }