今天我们继续来学习Java并发编程 Juc框架 ，把剩余部分学习完

  • 17万字 JUC 看这一篇就够了(一) (精华)
  • 17万字 JUC 看这一篇就够了(二) (精华)

文章目录

          • 非公原理
            • 加锁
            • 解锁
          • 公平原理
        • 可重入
        • 可打断
          • 基本使用
          • 实现原理
        • 锁超时
          • 基本使用
          • 实现原理
          • 哲学家就餐
        • 条件变量
          • 基本使用
          • 实现原理
            • await
            • signal
      • ReadWrite
        • 读写锁
        • 缓存应用
        • 实现原理
          • 成员属性
          • 加锁原理
          • 解锁原理
        • Stamped
      • CountDown
        • 基本使用
        • 实现原理
      • CyclicBarrier
        • 基本使用
        • 实现原理
          • 成员属性
          • 成员方法
      • Semaphore
        • 基本使用
        • 实现原理
        • PROPAGATE
      • Exchanger
    • 并发包
      • ConHashMap
        • 并发集合
          • 集合对比
          • 并发死链
        • 成员属性
          • 变量
          • 内部类
          • 代码块
        • 构造方法
        • 成员方法
          • 数据访存
          • 添加方法
          • 扩容方法
          • 获取方法
          • 删除方法
        • JDK7原理
      • CopyOnWrite
        • 原理分析
        • 弱一致性
        • 安全失败
      • Collections
      • SkipListMap
        • 底层结构
        • 成员变量
        • 成员方法
          • 其他方法
          • 添加方法
          • 获取方法
          • 删除方法
      • NoBlocking
        • 非阻塞队列
        • 构造方法
        • 入队方法
        • 出队方法
        • 成员方法
  • NET
    • DES
      • 网络编程
      • 通信协议
      • Java模型
    • I/O
      • IO模型
        • 五种模型
        • 阻塞式IO
        • 非阻塞式
        • 信号驱动
        • IO 复用
        • 异步 IO
      • 多路复用
        • select
          • 函数
          • 流程
        • poll
        • epoll
          • 函数
          • 特点
        • 应用
      • 系统调用
        • 内核态
        • 80中断
      • 零拷贝
        • DMA
        • BIO
        • mmap
        • sendfile
    • BIO
      • Inet
      • UDP
        • 基本介绍
        • 实现UDP
        • 通讯方式
      • TCP
        • 基本介绍
        • Socket
        • 实现TCP
          • 开发流程
          • 实现通信
          • 伪异步
        • 文件传输
          • 字节流
          • 数据流
    • NIO
      • 基本介绍
      • 实现原理
      • 缓冲区
        • 基本介绍
        • 基本属性
        • 常用API
        • 读写数据
        • 粘包拆包
      • 直接内存
        • 基本介绍
        • 通信原理
        • 分配回收
        • 共享内存
      • 通道
        • 基本介绍
        • 常用API
        • 文件读写
        • 文件复制
        • 分散聚集
      • 选择器
        • 基本介绍
        • 常用API
      • NIO实现
        • 常用API
        • 代码实现
    • AIO
非公原理
加锁

NonfairSync 继承自 AQS

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

  • 没有竞争：ExclusiveOwnerThread 属于 Thread-0，state 设置为 1

    // ReentrantLock.NonfairSync#lock
final void lock() {// 用 cas 尝试（仅尝试一次）将 state 从 0 改为 1, 如果成功表示【获得了独占锁】if (compareAndSetState(0, 1))// 设置当前线程为独占线程setExclusiveOwnerThread(Thread.currentThread());elseacquire(1);//失败进入
}

  • 第一个竞争出现：Thread-1 执行，CAS 尝试将 state 由 0 改为 1，结果失败（第一次），进入 acquire 逻辑

    // AbstractQueuedSynchronizer#acquire
public final void acquire(int arg) {// tryAcquire 尝试获取锁失败时, 会调用 addWaiter 将当前线程封装成node入队，acquireQueued 阻塞当前线程，// acquireQueued 返回 true 表示挂起过程中线程被中断唤醒过，false 表示未被中断过if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg))// 如果线程被中断了逻辑来到这，完成一次真正的打断效果selfInterrupt();
}

  • 进入 tryAcquire 尝试获取锁逻辑，这时 state 已经是1，结果仍然失败（第二次），加锁成功有两种情况：

    • 当前 AQS 处于无锁状态
    • 加锁线程就是当前线程，说明发生了锁重入
    // ReentrantLock.NonfairSync#tryAcquire
protected final boolean tryAcquire(int acquires) {return nonfairTryAcquire(acquires);
}
// 抢占成功返回 true，抢占失败返回 false
final boolean nonfairTryAcquire(int acquires) {final Thread current = Thread.currentThread();// state 值int c = getState();// 条件成立说明当前处于【无锁状态】if (c == 0) {//如果还没有获得锁，尝试用cas获得，这里体现非公平性: 不去检查 AQS 队列是否有阻塞线程直接获取锁        if (compareAndSetState(0, acquires)) {// 获取锁成功设置当前线程为独占锁线程。setExclusiveOwnerThread(current);return true;}    }    // 如果已经有线程获得了锁, 独占锁线程还是当前线程, 表示【发生了锁重入】else if (current == getExclusiveOwnerThread()) {// 更新锁重入的值int nextc = c + acquires;// 越界判断，当重入的深度很深时，会导致 nextc < 0，int值达到最大之后再 + 1 变负数if (nextc < 0) // overflowthrow new Error("Maximum lock count exceeded");// 更新 state 的值，这里不使用 cas 是因为当前线程正在持有锁，所以这里的操作相当于在一个管程内setState(nextc);return true;}// 获取失败return false;
}

  • 接下来进入 addWaiter 逻辑，构造 Node 队列（不是阻塞队列），前置条件是当前线程获取锁失败，说明有线程占用了锁

    • 图中黄色三角表示该 Node 的 waitStatus 状态，其中 0 为默认正常状态
    • Node 的创建是懒惰的，其中第一个 Node 称为 Dummy（哑元）或哨兵，用来占位，并不关联线程
    // AbstractQueuedSynchronizer#addWaiter，返回当前线程的 node 节点
private Node addWaiter(Node mode) {// 将当前线程关联到一个 Node 对象上, 模式为独占模式   Node node = new Node(Thread.currentThread(), mode);Node pred = tail;// 快速入队，如果 tail 不为 null，说明存在队列if (pred != null) {// 将当前节点的前驱节点指向 尾节点node.prev = pred;// 通过 cas 将 Node 对象加入 AQS 队列，成为尾节点，【尾插法】if (compareAndSetTail(pred, node)) {pred.next = node;// 双向链表return node;}}// 初始时队列为空，或者 CAS 失败进入这里enq(node);return node;
}

    // AbstractQueuedSynchronizer#enq
private Node enq(final Node node) {// 自旋入队，必须入队成功才结束循环for (;;) {Node t = tail;// 说明当前锁被占用，且当前线程可能是【第一个获取锁失败】的线程，【还没有建立队列】if (t == null) {// 设置一个【哑元节点】，头尾指针都指向该节点if (compareAndSetHead(new Node()))tail = head;} else {// 自旋到这，普通入队方式，首先赋值尾节点的前驱节点【尾插法】node.prev = t;// 【在设置完尾节点后，才更新的原始尾节点的后继节点，所以此时从前往后遍历会丢失尾节点】if (compareAndSetTail(t, node)) {//【此时 t.next  = null，并且这里已经 CAS 结束，线程并不是安全的】t.next = node;return t; // 返回当前 node 的前驱节点}}}
}

  • 线程节点加入队列成功，进入 AbstractQueuedSynchronizer#acquireQueued 逻辑阻塞线程

    • acquireQueued 会在一个自旋中不断尝试获得锁，失败后进入 park 阻塞

    • 如果当前线程是在 head 节点后，会再次 tryAcquire 尝试获取锁，state 仍为 1 则失败（第三次）

    final boolean acquireQueued(final Node node, int arg) {// true 表示当前线程抢占锁失败，false 表示成功boolean failed = true;try {// 中断标记，表示当前线程是否被中断boolean interrupted = false;for (;;) {// 获得当前线程节点的前驱节点final Node p = node.predecessor();// 前驱节点是 head, FIFO 队列的特性表示轮到当前线程可以去获取锁if (p == head && tryAcquire(arg)) {// 获取成功, 设置当前线程自己的 node 为 headsetHead(node);p.next = null; // help GC// 表示抢占锁成功failed = false;// 返回当前线程是否被中断return interrupted;}// 判断是否应当 park，返回 false 后需要新一轮的循环，返回 true 进入条件二阻塞线程if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())// 条件二返回结果是当前线程是否被打断，没有被打断返回 false 不进入这里的逻辑// 【就算被打断了，也会继续循环，并不会返回】interrupted = true;}} finally {// 【可打断模式下才会进入该逻辑】if (failed)cancelAcquire(node);}
}
    • 进入 shouldParkAfterFailedAcquire 逻辑，将前驱 node 的 waitStatus 改为 -1，返回 false；waitStatus 为 -1 的节点用来唤醒下一个节点
    private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {int ws = pred.waitStatus;// 表示前置节点是个可以唤醒当前节点的节点，返回 trueif (ws == Node.SIGNAL)return true;// 前置节点的状态处于取消状态，需要【删除前面所有取消的节点】, 返回到外层循环重试if (ws > 0) {do {node.prev = pred = pred.prev;} while (pred.waitStatus > 0);// 获取到非取消的节点，连接上当前节点pred.next = node;// 默认情况下 node 的 waitStatus 是 0，进入这里的逻辑} else {// 【设置上一个节点状态为 Node.SIGNAL】，返回外层循环重试compareAndSetWaitStatus(pred, ws, Node.SIGNAL);}// 返回不应该 park，再次尝试一次return false;
}
    • shouldParkAfterFailedAcquire 执行完毕回到 acquireQueued ，再次 tryAcquire 尝试获取锁，这时 state 仍为 1 获取失败（第四次）
    • 当再次进入 shouldParkAfterFailedAcquire 时，这时其前驱 node 的 waitStatus 已经是 -1 了，返回 true
    • 进入 parkAndCheckInterrupt， Thread-1 park（灰色表示）
    private final boolean parkAndCheckInterrupt() {// 阻塞当前线程，如果打断标记已经是 true, 则 park 会失效LockSupport.park(this);// 判断当前线程是否被打断，清除打断标记return Thread.interrupted();
}

  • 再有多个线程经历竞争失败后：

    [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-tQ7iZKgb-1679358011068)(https://seazean.oss-cn-beijing.aliyuncs.com/img/Java/JUC-ReentrantLock-非公平锁3.png)]

解锁

ReentrantLock#unlock：释放锁

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

Thread-0 释放锁，进入 release 流程

  • 进入 tryRelease，设置 exclusiveOwnerThread 为 null，state = 0

  • 当前队列不为 null，并且 head 的 waitStatus = -1，进入 unparkSuccessor

    // AbstractQueuedSynchronizer#release
public final boolean release(int arg) {// 尝试释放锁，tryRelease 返回 true 表示当前线程已经【完全释放锁，重入的释放了】if (tryRelease(arg)) {// 队列头节点Node h = head;// 头节点什么时候是空？没有发生锁竞争，没有竞争线程创建哑元节点// 条件成立说明阻塞队列有等待线程，需要唤醒 head 节点后面的线程if (h != null && h.waitStatus != 0)unparkSuccessor(h);return true;}    return false;
}

    // ReentrantLock.Sync#tryRelease
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);}// 当前线程就是持有锁线程，所以可以直接更新锁，不需要使用 CASsetState(c);return free;
}

  • 进入 AbstractQueuedSynchronizer#unparkSuccessor 方法，唤醒当前节点的后继节点

    • 找到队列中距离 head 最近的一个没取消的 Node，unpark 恢复其运行，本例中即为 Thread-1
    • 回到 Thread-1 的 acquireQueued 流程
    private void unparkSuccessor(Node node) {// 当前节点的状态int ws = node.waitStatus;    if (ws < 0)        // 【尝试重置状态为 0】，因为当前节点要完成对后续节点的唤醒任务了，不需要 -1 了compareAndSetWaitStatus(node, ws, 0);    // 找到需要 unpark 的节点，当前节点的下一个    Node s = node.next;    // 已取消的节点不能唤醒，需要找到距离头节点最近的非取消的节点if (s == null || s.waitStatus > 0) {s = null;// AQS 队列【从后至前】找需要 unpark 的节点，直到 t == 当前的 node 为止，找不到就不唤醒了for (Node t = tail; t != null && t != node; t = t.prev)// 说明当前线程状态需要被唤醒if (t.waitStatus <= 0)// 置换引用s = t;}// 【找到合适的可以被唤醒的 node，则唤醒线程】if (s != null)LockSupport.unpark(s.thread);
}

    从后向前的唤醒的原因：enq 方法中，节点是尾插法，首先赋值的是尾节点的前驱节点，此时前驱节点的 next 并没有指向尾节点，从前遍历会丢失尾节点

  • 唤醒的线程会从 park 位置开始执行，如果加锁成功（没有竞争），会设置

    • exclusiveOwnerThread 为 Thread-1，state = 1
    • head 指向刚刚 Thread-1 所在的 Node，该 Node 会清空 Thread
    • 原本的 head 因为从链表断开，而可被垃圾回收（图中有错误，原来的头节点的 waitStatus 被改为 0 了）

    [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-LXCp0lox-1679358011068)(https://seazean.oss-cn-beijing.aliyuncs.com/img/Java/JUC-ReentrantLock-非公平锁4.png)]

  • 如果这时有其它线程来竞争**（非公平）**，例如这时有 Thread-4 来了并抢占了锁

    • Thread-4 被设置为 exclusiveOwnerThread，state = 1
    • Thread-1 再次进入 acquireQueued 流程，获取锁失败，重新进入 park 阻塞

    [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-0eQ8vo9D-1679358011069)(https://seazean.oss-cn-beijing.aliyuncs.com/img/Java/JUC-ReentrantLock-非公平锁5.png)]

公平原理

与非公平锁主要区别在于 tryAcquire 方法：先检查 AQS 队列中是否有前驱节点，没有才去 CAS 竞争

static final class FairSync extends Sync {private static final long serialVersionUID = -3000897897090466540L;final void lock() {acquire(1);}protected final boolean tryAcquire(int acquires) {final Thread current = Thread.currentThread();int c = getState();if (c == 0) {// 先检查 AQS 队列中是否有前驱节点, 没有(false)才去竞争if (!hasQueuedPredecessors() &&compareAndSetState(0, acquires)) {setExclusiveOwnerThread(current);return true;}}// 锁重入return false;}
}

public final boolean hasQueuedPredecessors() {    Node t = tail;Node h = head;Node s;    // 头尾指向一个节点，链表为空，返回falsereturn h != t &&// 头尾之间有节点，判断头节点的下一个是不是空// 不是空进入最后的判断，第二个节点的线程是否是本线程，不是返回 true，表示当前节点有前驱节点((s = h.next) == null || s.thread != Thread.currentThread());
}

可重入

可重入是指同一个线程如果首次获得了这把锁，那么它是这把锁的拥有者，因此有权利再次获取这把锁，如果不可重入锁，那么第二次获得锁时，自己也会被锁挡住，直接造成死锁

源码解析参考：nonfairTryAcquire(int acquires)) 和 tryRelease(int releases)

static ReentrantLock lock = new ReentrantLock();
public static void main(String[] args) {method1();
}
public static void method1() {lock.lock();try {System.out.println(Thread.currentThread().getName() + " execute method1");method2();} finally {lock.unlock();}
}
public static void method2() {lock.lock();try {System.out.println(Thread.currentThread().getName() + " execute method2");} finally {lock.unlock();}
}

在 Lock 方法加两把锁会是什么情况呢？

  • 加锁两次解锁两次：正常执行
  • 加锁两次解锁一次：程序直接卡死，线程不能出来，也就说明申请几把锁，最后需要解除几把锁
  • 加锁一次解锁两次：运行程序会直接报错
public void getLock() {lock.lock();lock.lock();try {System.out.println(Thread.currentThread().getName() + "\t get Lock");} finally {lock.unlock();//lock.unlock();}
}

可打断

基本使用

public void lockInterruptibly()：获得可打断的锁

  • 如果没有竞争此方法就会获取 lock 对象锁
  • 如果有竞争就进入阻塞队列，可以被其他线程用 interrupt 打断

注意：如果是不可中断模式，那么即使使用了 interrupt 也不会让等待状态中的线程中断

public static void main(String[] args) throws InterruptedException {    ReentrantLock lock = new ReentrantLock();    Thread t1 = new Thread(() -> {        try {            System.out.println("尝试获取锁");            lock.lockInterruptibly();        } catch (InterruptedException e) {            System.out.println("没有获取到锁，被打断，直接返回");            return;        }        try {            System.out.println("获取到锁");        } finally {            lock.unlock();        }    }, "t1");    lock.lock();    t1.start();    Thread.sleep(2000);    System.out.println("主线程进行打断锁");    t1.interrupt();
}
实现原理

  • 不可打断模式：即使它被打断，仍会驻留在 AQS 阻塞队列中，一直要等到获得锁后才能得知自己被打断了

    public final void acquire(int arg) {    if (!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg))//阻塞等待        // 如果acquireQueued返回true，打断状态 interrupted = true        selfInterrupt();
}
static void selfInterrupt() {// 知道自己被打断了，需要重新产生一次中断完成中断效果Thread.currentThread().interrupt();
}

    final boolean acquireQueued(final Node node, int arg) {    try {        boolean interrupted = false;        for (;;) {            final Node p = node.predecessor();            if (p == head && tryAcquire(arg)) {                setHead(node);                p.next = null; // help GC                failed = false;                // 还是需要获得锁后, 才能返回打断状态return interrupted;            }            if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()){// 条件二中判断当前线程是否被打断，被打断返回true，设置中断标记为 true，【获取锁后返回】interrupted = true;  }                  } } finally {if (failed)cancelAcquire(node);}
}private final boolean parkAndCheckInterrupt() {    // 阻塞当前线程，如果打断标记已经是 true, 则 park 会失效LockSupport.park(this);    // 判断当前线程是否被打断，清除打断标记，被打断返回truereturn Thread.interrupted();}

  • 可打断模式：AbstractQueuedSynchronizer#acquireInterruptibly，被打断后会直接抛出异常

    public void lockInterruptibly() throws InterruptedException {    sync.acquireInterruptibly(1);
}
public final void acquireInterruptibly(int arg) {// 被其他线程打断了直接返回 falseif (Thread.interrupted())throw new InterruptedException();if (!tryAcquire(arg))// 没获取到锁，进入这里doAcquireInterruptibly(arg);
}

    private void doAcquireInterruptibly(int arg) throws InterruptedException {// 返回封装当前线程的节点final Node node = addWaiter(Node.EXCLUSIVE);boolean failed = true;try {for (;;) {//...if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt())// 【在 park 过程中如果被 interrupt 会抛出异常】, 而不会再次进入循环获取锁后才完成打断效果throw new InterruptedException();}    } finally {// 抛出异常前会进入这里if (failed)// 取消当前线程的节点cancelAcquire(node);}
}

    // 取消节点出队的逻辑
private void cancelAcquire(Node node) {// 判空if (node == null)return;// 把当前节点封装的 Thread 置为空node.thread = null;// 获取当前取消的 node 的前驱节点Node pred = node.prev;// 前驱节点也被取消了，循环找到前面最近的没被取消的节点while (pred.waitStatus > 0)node.prev = pred = pred.prev;// 获取前驱节点的后继节点，可能是当前 node，也可能是 waitStatus > 0 的节点Node predNext = pred.next;// 把当前节点的状态设置为 【取消状态 1】node.waitStatus = Node.CANCELLED;// 条件成立说明当前节点是尾节点，把当前节点的前驱节点设置为尾节点if (node == tail && compareAndSetTail(node, pred)) {// 把前驱节点的后继节点置空，这里直接把所有的取消节点出队compareAndSetNext(pred, predNext, null);} else {// 说明当前节点不是 tail 节点int ws;// 条件一成立说明当前节点不是 head.next 节点if (pred != head &&// 判断前驱节点的状态是不是 -1，不成立说明前驱状态可能是 0 或者刚被其他线程取消排队了((ws = pred.waitStatus) == Node.SIGNAL ||// 如果状态不是 -1，设置前驱节点的状态为 -1(ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) &&// 前驱节点的线程不为nullpred.thread != null) {Node next = node.next;// 当前节点的后继节点是正常节点if (next != null && next.waitStatus <= 0)// 把 前驱节点的后继节点 设置为 当前节点的后继节点，【从队列中删除了当前节点】compareAndSetNext(pred, predNext, next);} else {// 当前节点是 head.next 节点，唤醒当前节点的后继节点unparkSuccessor(node);}node.next = node; // help GC}
}

锁超时

基本使用

public boolean tryLock()：尝试获取锁，获取到返回 true，获取不到直接放弃，不进入阻塞队列

public boolean tryLock(long timeout, TimeUnit unit)：在给定时间内获取锁，获取不到就退出

注意：tryLock 期间也可以被打断

public static void main(String[] args) {ReentrantLock lock = new ReentrantLock();Thread t1 = new Thread(() -> {try {if (!lock.tryLock(2, TimeUnit.SECONDS)) {System.out.println("获取不到锁");return;}} catch (InterruptedException e) {System.out.println("被打断，获取不到锁");return;}try {log.debug("获取到锁");} finally {lock.unlock();}}, "t1");lock.lock();System.out.println("主线程获取到锁");t1.start();Thread.sleep(1000);try {System.out.println("主线程释放了锁");} finally {lock.unlock();}
}
实现原理

  • 成员变量：指定超时限制的阈值，小于该值的线程不会被挂起

    static final long spinForTimeoutThreshold = 1000L;

    超时时间设置的小于该值，就会被禁止挂起，因为阻塞在唤醒的成本太高，不如选择自旋空转

  • tryLock()

    public boolean tryLock() {   // 只尝试一次return sync.nonfairTryAcquire(1);
}

  • tryLock(long timeout, TimeUnit unit)

    public final boolean tryAcquireNanos(int arg, long nanosTimeout) {if (Thread.interrupted())        throw new InterruptedException();    // tryAcquire 尝试一次return tryAcquire(arg) || doAcquireNanos(arg, nanosTimeout);
}
protected final boolean tryAcquire(int acquires) {    return nonfairTryAcquire(acquires);
}

    private boolean doAcquireNanos(int arg, long nanosTimeout) {    if (nanosTimeout <= 0L)return false;// 获取最后期限的时间戳final long deadline = System.nanoTime() + nanosTimeout;//...try {for (;;) {//...// 计算还需等待的时间nanosTimeout = deadline - System.nanoTime();if (nanosTimeout <= 0L)    //时间已到     return false;if (shouldParkAfterFailedAcquire(p, node) &&// 如果 nanosTimeout 大于该值，才有阻塞的意义，否则直接自旋会好点nanosTimeout > spinForTimeoutThreshold)LockSupport.parkNanos(this, nanosTimeout);// 【被打断会报异常】if (Thread.interrupted())throw new InterruptedException();}    }
}
哲学家就餐
public static void main(String[] args) {Chopstick c1 = new Chopstick("1");//...Chopstick c5 = new Chopstick("5");new Philosopher("苏格拉底", c1, c2).start();new Philosopher("柏拉图", c2, c3).start();new Philosopher("亚里士多德", c3, c4).start();new Philosopher("赫拉克利特", c4, c5).start();    new Philosopher("阿基米德", c5, c1).start();
}
class Philosopher extends Thread {Chopstick left;Chopstick right;public void run() {while (true) {// 尝试获得左手筷子if (left.tryLock()) {try {// 尝试获得右手筷子if (right.tryLock()) {try {System.out.println("eating...");Thread.sleep(1000);} finally {right.unlock();}}} finally {left.unlock();}}}}
}
class Chopstick extends ReentrantLock {String name;public Chopstick(String name) {this.name = name;}@Overridepublic String toString() {return "筷子{" + name + '}';}
}

条件变量

基本使用

synchronized 的条件变量，是当条件不满足时进入 WaitSet 等待；ReentrantLock 的条件变量比 synchronized 强大之处在于支持多个条件变量

ReentrantLock 类获取 Condition 对象：public Condition newCondition()

Condition 类 API：

  • void await()：当前线程从运行状态进入等待状态，释放锁
  • void signal()：唤醒一个等待在 Condition 上的线程，但是必须获得与该 Condition 相关的锁

使用流程：

  • await / signal 前需要获得锁

  • await 执行后，会释放锁进入 ConditionObject 等待

  • await 的线程被唤醒去重新竞争 lock 锁

  • 线程在条件队列被打断会抛出中断异常

  • 竞争 lock 锁成功后，从 await 后继续执行

public static void main(String[] args) throws InterruptedException {    ReentrantLock lock = new ReentrantLock();//创建一个新的条件变量Condition condition1 = lock.newCondition();Condition condition2 = lock.newCondition();new Thread(() -> {try {lock.lock();System.out.println("进入等待");//进入休息室等待condition1.await();System.out.println("被唤醒了");} catch (InterruptedException e) {e.printStackTrace();} finally {lock.unlock();}    }).start();Thread.sleep(1000);//叫醒new Thread(() -> {try {            lock.lock();//唤醒condition2.signal();} finally {lock.unlock();}}).start();
}
实现原理
await

总体流程是将 await 线程包装成 node 节点放入 ConditionObject 的条件队列，如果被唤醒就将 node 转移到 AQS 的执行阻塞队列，等待获取锁，每个 Condition 对象都包含一个等待队列

  • 开始 Thread-0 持有锁，调用 await，线程进入 ConditionObject 等待，直到被唤醒或打断，调用 await 方法的线程都是持锁状态的，所以说逻辑里不存在并发

    public final void await() throws InterruptedException {// 判断当前线程是否是中断状态，是就直接给个中断异常if (Thread.interrupted())throw new InterruptedException();// 将调用 await 的线程包装成 Node，添加到条件队列并返回Node node = addConditionWaiter();// 完全释放节点持有的锁，因为其他线程唤醒当前线程的前提是【持有锁】int savedState = fullyRelease(node);// 设置打断模式为没有被打断，状态码为 0int interruptMode = 0;// 如果该节点还没有转移至 AQS 阻塞队列, park 阻塞，等待进入阻塞队列while (!isOnSyncQueue(node)) {LockSupport.park(this);// 如果被打断，退出等待队列，对应的 node 【也会被迁移到阻塞队列】尾部，状态设置为 0if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)break;}// 逻辑到这说明当前线程退出等待队列，进入【阻塞队列】// 尝试枪锁，释放了多少锁就【重新获取多少锁】，获取锁成功判断打断模式if (acquireQueued(node, savedState) && interruptMode != THROW_IE)interruptMode = REINTERRUPT;// node 在条件队列时 如果被外部线程中断唤醒，会加入到阻塞队列，但是并未设 nextWaiter = nullif (node.nextWaiter != null)// 清理条件队列内所有已取消的 NodeunlinkCancelledWaiters();// 条件成立说明挂起期间发生过中断if (interruptMode != 0)// 应用打断模式reportInterruptAfterWait(interruptMode);
}

    // 打断模式 - 在退出等待时重新设置打断状态
private static final int REINTERRUPT = 1;
// 打断模式 - 在退出等待时抛出异常
private static final int THROW_IE = -1;

    [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-0V5nt2PN-1679358011069)(https://seazean.oss-cn-beijing.aliyuncs.com/img/Java/JUC-ReentrantLock-条件变量1.png)]

  • 创建新的 Node 状态为 -2（Node.CONDITION），关联 Thread-0，加入等待队列尾部

    private Node addConditionWaiter() {// 获取当前条件队列的尾节点的引用，保存到局部变量 t 中Node t = lastWaiter;// 当前队列中不是空，并且节点的状态不是 CONDITION（-2），说明当前节点发生了中断if (t != null && t.waitStatus != Node.CONDITION) {// 清理条件队列内所有已取消的 NodeunlinkCancelledWaiters();// 清理完成重新获取 尾节点 的引用t = lastWaiter;}// 创建一个关联当前线程的新 node, 设置状态为 CONDITION(-2)，添加至队列尾部Node node = new Node(Thread.currentThread(), Node.CONDITION);if (t == null)firstWaiter = node;      // 空队列直接放在队首【不用CAS因为执行线程是持锁线程，并发安全】elset.nextWaiter = node; // 非空队列队尾追加lastWaiter = node;          // 更新队尾的引用return node;
}

    // 清理条件队列内所有已取消（不是CONDITION）的 node，【链表删除的逻辑】
private void unlinkCancelledWaiters() {// 从头节点开始遍历【FIFO】Node t = firstWaiter;// 指向正常的 CONDITION 节点Node trail = null;// 等待队列不空while (t != null) {// 获取当前节点的后继节点Node next = t.nextWaiter;// 判断 t 节点是不是 CONDITION 节点，条件队列内不是 CONDITION 就不是正常的if (t.waitStatus != Node.CONDITION) { // 不是正常节点，需要 t 与下一个节点断开t.nextWaiter = null;// 条件成立说明遍历到的节点还未碰到过正常节点if (trail == null)// 更新 firstWaiter 指针为下个节点firstWaiter = next;else// 让上一个正常节点指向 当前取消节点的 下一个节点，【删除非正常的节点】trail.nextWaiter = next;// t 是尾节点了，更新 lastWaiter 指向最后一个正常节点if (next == null)lastWaiter = trail;} else {// trail 指向的是正常节点 trail = t;}// 把 t.next 赋值给 t，循环遍历t = next; }
}

  • 接下来 Thread-0 进入 AQS 的 fullyRelease 流程，释放同步器上的锁

    // 线程可能重入，需要将 state 全部释放
final int fullyRelease(Node node) {// 完全释放锁是否成功，false 代表成功boolean failed = true;try {// 获取当前线程所持有的 state 值总数int savedState = getState();// release -> tryRelease 解锁重入锁if (release(savedState)) {// 释放成功failed = false;// 返回解锁的深度return savedState;} else {// 解锁失败抛出异常throw new IllegalMonitorStateException();}} finally {// 没有释放成功，将当前 node 设置为取消状态if (failed)node.waitStatus = Node.CANCELLED;}
}

  • fullyRelease 中会 unpark AQS 队列中的下一个节点竞争锁，假设 Thread-1 竞争成功

    [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-4pT5CYeh-1679358011069)(https://seazean.oss-cn-beijing.aliyuncs.com/img/Java/JUC-ReentrantLock-条件变量2.png)]

  • Thread-0 进入 isOnSyncQueue 逻辑判断节点是否移动到阻塞队列，没有就 park 阻塞 Thread-0

    final boolean isOnSyncQueue(Node node) {// node 的状态是 CONDITION，signal 方法是先修改状态再迁移，所以前驱节点为空证明还【没有完成迁移】if (node.waitStatus == Node.CONDITION || node.prev == null)return false;// 说明当前节点已经成功入队到阻塞队列，且当前节点后面已经有其它 node，因为条件队列的 next 指针为 nullif (node.next != null)return true;// 说明【可能在阻塞队列，但是是尾节点】// 从阻塞队列的尾节点开始向前【遍历查找 node】，如果查找到返回 true，查找不到返回 falsereturn findNodeFromTail(node);
}

  • await 线程 park 后如果被 unpark 或者被打断，都会进入 checkInterruptWhileWaiting 判断线程是否被打断：在条件队列被打断的线程需要抛出异常

    private int checkInterruptWhileWaiting(Node node) {// Thread.interrupted() 返回当前线程中断标记位，并且重置当前标记位 为 false// 如果被中断了，根据是否在条件队列被中断的，设置中断状态码return Thread.interrupted() ?(transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) : 0;
}

    // 这个方法只有在线程是被打断唤醒时才会调用
final boolean transferAfterCancelledWait(Node node) {// 条件成立说明当前node一定是在条件队列内，因为 signal 迁移节点到阻塞队列时，会将节点的状态修改为 0if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {// 把【中断唤醒的 node 加入到阻塞队列中】enq(node);// 表示是在条件队列内被中断了，设置为 THROW_IE 为 -1return true;}//执行到这里的情况：//1.当前node已经被外部线程调用 signal 方法将其迁移到 阻塞队列 内了//2.当前node正在被外部线程调用 signal 方法将其迁移至 阻塞队列 进行中状态// 如果当前线程还没到阻塞队列，一直释放 CPUwhile (!isOnSyncQueue(node))Thread.yield();// 表示当前节点被中断唤醒时不在条件队列了，设置为 REINTERRUPT 为 1return false;
}

  • 最后开始处理中断状态：

    private void reportInterruptAfterWait(int interruptMode) throws InterruptedException {// 条件成立说明【在条件队列内发生过中断，此时 await 方法抛出中断异常】if (interruptMode == THROW_IE)throw new InterruptedException();// 条件成立说明【在条件队列外发生的中断，此时设置当前线程的中断标记位为 true】else if (interruptMode == REINTERRUPT)// 进行一次自己打断，产生中断的效果selfInterrupt();
}
signal

  • 假设 Thread-1 要来唤醒 Thread-0，进入 ConditionObject 的 doSignal 流程，取得等待队列中第一个 Node，即 Thread-0 所在 Node，必须持有锁才能唤醒, 因此 doSignal 内线程安全

    public final void signal() {// 判断调用 signal 方法的线程是否是独占锁持有线程if (!isHeldExclusively())throw new IllegalMonitorStateException();// 获取条件队列中第一个 NodeNode first = firstWaiter;// 不为空就将第该节点【迁移到阻塞队列】if (first != null)doSignal(first);
}

    // 唤醒 - 【将没取消的第一个节点转移至 AQS 队列尾部】
private void doSignal(Node first) {do {// 成立说明当前节点的下一个节点是 null，当前节点是尾节点了，队列中只有当前一个节点了if ((firstWaiter = first.nextWaiter) == null)lastWaiter = null;first.nextWaiter = null;// 将等待队列中的 Node 转移至 AQS 队列，不成功且还有节点则继续循环} while (!transferForSignal(first) && (first = firstWaiter) != null);
}// signalAll() 会调用这个函数，唤醒所有的节点
private void doSignalAll(Node first) {lastWaiter = firstWaiter = null;do {Node next = first.nextWaiter;first.nextWaiter = null;transferForSignal(first);first = next;// 唤醒所有的节点，都放到阻塞队列中} while (first != null);
}

  • 执行 transferForSignal，先将节点的 waitStatus 改为 0，然后加入 AQS 阻塞队列尾部，将 Thread-3 的 waitStatus 改为 -1

    // 如果节点状态是取消, 返回 false 表示转移失败, 否则转移成功
final boolean transferForSignal(Node node) {// CAS 修改当前节点的状态，修改为 0，因为当前节点马上要迁移到阻塞队列了// 如果状态已经不是 CONDITION, 说明线程被取消（await 释放全部锁失败）或者被中断（可打断 cancelAcquire）if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))// 返回函数调用处继续寻找下一个节点return false;// 【先改状态，再进行迁移】// 将当前 node 入阻塞队列，p 是当前节点在阻塞队列的【前驱节点】Node p = enq(node);int ws = p.waitStatus;// 如果前驱节点被取消或者不能设置状态为 Node.SIGNAL，就 unpark 取消当前节点线程的阻塞状态, // 让 thread-0 线程竞争锁，重新同步状态if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))LockSupport.unpark(node.thread);return true;
}

    [外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-tTefvvSx-1679358011070)(https://seazean.oss-cn-beijing.aliyuncs.com/img/Java/JUC-ReentrantLock-条件变量3.png)]

  • Thread-1 释放锁，进入 unlock 流程

ReadWrite

读写锁

独占锁：指该锁一次只能被一个线程所持有，对 ReentrantLock 和 Synchronized 而言都是独占锁

共享锁：指该锁可以被多个线程锁持有

ReentrantReadWriteLock 其读锁是共享锁，写锁是独占锁

作用：多个线程同时读一个资源类没有任何问题，为了满足并发量，读取共享资源应该同时进行，但是如果一个线程想去写共享资源，就不应该再有其它线程可以对该资源进行读或写

使用规则：

  • 加锁解锁格式：

    r.lock();
try {// 临界区
} finally {r.unlock();
}

  • 读-读能共存、读-写不能共存、写-写不能共存

  • 读锁不支持条件变量

  • 重入时升级不支持：持有读锁的情况下去获取写锁会导致获取写锁永久等待，需要先释放读，再去获得写

  • 重入时降级支持：持有写锁的情况下去获取读锁，造成只有当前线程会持有读锁，因为写锁会互斥其他的锁

    w.lock();
try {r.lock();// 降级为读锁, 释放写锁, 这样能够让其它线程读取缓存try {// ...} finally{w.unlock();// 要在写锁释放之前获取读锁}
} finally{r.unlock();
}

构造方法：

  • public ReentrantReadWriteLock()：默认构造方法，非公平锁
  • public ReentrantReadWriteLock(boolean fair)：true 为公平锁

常用API：

  • public ReentrantReadWriteLock.ReadLock readLock()：返回读锁
  • public ReentrantReadWriteLock.WriteLock writeLock()：返回写锁
  • public void lock()：加锁
  • public void unlock()：解锁
  • public boolean tryLock()：尝试获取锁

读读并发：

public static void main(String[] args) {ReentrantReadWriteLock rw = new ReentrantReadWriteLock();ReentrantReadWriteLock.ReadLock r = rw.readLock();ReentrantReadWriteLock.WriteLock w = rw.writeLock();new Thread(() -> {r.lock();try {Thread.sleep(2000);System.out.println("Thread 1 running " + new Date());} finally {r.unlock();}},"t1").start();new Thread(() -> {r.lock();try {Thread.sleep(2000);System.out.println("Thread 2 running " + new Date());} finally {r.unlock();}},"t2").start();
}

缓存应用

缓存更新时，是先清缓存还是先更新数据库

  • 先清缓存：可能造成刚清理缓存还没有更新数据库，线程直接查询了数据库更新过期数据到缓存

  • 先更新据库：可能造成刚更新数据库，还没清空缓存就有线程从缓存拿到了旧数据

  • 补充情况：查询线程 A 查询数据时恰好缓存数据由于时间到期失效，或是第一次查询

可以使用读写锁进行操作

实现原理

成员属性

读写锁用的是同一个 Sycn 同步器，因此等待队列、state 等也是同一个，原理与 ReentrantLock 加锁相比没有特殊之处，不同是写锁状态占了 state 的低 16 位，而读锁使用的是 state 的高 16 位

  • 读写锁：

    private final ReentrantReadWriteLock.ReadLock readerLock;
private final ReentrantReadWriteLock.WriteLock writerLock;

  • 构造方法：默认是非公平锁，可以指定参数创建公平锁

    public ReentrantReadWriteLock(boolean fair) {// true 为公平锁sync = fair ? new FairSync() : new NonfairSync();// 这两个 lock 共享同一个 sync 实例，都是由 ReentrantReadWriteLock 的 sync 提供同步实现readerLock = new ReadLock(this);writerLock = new WriteLock(this);
}

Sync 类的属性：

  • 统计变量：

    // 用来移位
static final int SHARED_SHIFT   = 16;
// 高16位的1
static final int SHARED_UNIT    = (1 << SHARED_SHIFT);
// 65535，16个1，代表写锁的最大重入次数
static final int MAX_COUNT      = (1 << SHARED_SHIFT) - 1;
// 低16位掩码：0b 1111 1111 1111 1111，用来获取写锁重入的次数
static final int EXCLUSIVE_MASK = (1 << SHARED_SHIFT) - 1;

  • 获取读写锁的次数：

    // 获取读写锁的读锁分配的总次数
static int sharedCount(int c)    { return c >>> SHARED_SHIFT; }
// 写锁（独占）锁的重入次数
static int exclusiveCount(int c) { return c & EXCLUSIVE_MASK; }

  • 内部类：

    // 记录读锁线程自己的持有读锁的数量（重入次数），因为 state 高16位记录的是全局范围内所有的读线程获取读锁的总量
static final class HoldCounter {int count = 0;// Use id, not reference, to avoid garbage retentionfinal long tid = getThreadId(Thread.currentThread());
}
// 线程安全的存放线程各自的 HoldCounter 对象
static final class ThreadLocalHoldCounter extends ThreadLocal<HoldCounter> {public HoldCounter initialValue() {return new HoldCounter();}
}

  • 内部类实例：

    // 当前线程持有的可重入读锁的数量，计数为 0 时删除
private transient ThreadLocalHoldCounter readHolds;
// 记录最后一个获取【读锁】线程的 HoldCounter 对象
private transient HoldCounter cachedHoldCounter;

  • 首次获取锁：

    // 第一个获取读锁的线程
private transient Thread firstReader = null;
// 记录该线程持有的读锁次数（读锁重入次数）
private transient int firstReaderHoldCount;

  • Sync 构造方法：

    Sync() {readHolds = new ThreadLocalHoldCounter();// 确保其他线程的数据可见性，state 是 volatile 修饰的变量，重写该值会将线程本地缓存数据【同步至主存】setState(getState());
}
加锁原理