1 简介

• JUC包简介

  • JUC全称java.util.concurrent，是JDK提供的并发工具包
  • JUC包由Doug Lea大师编写，提供了大量实用，高性能的工具类

• JUC包结构图

• JUC包下包含了阻塞队列、线程池、并发容器等，另外还包含了两个子包atomic和locks

• JUC包的整体实现图

• JUC包中类的实现主要是依赖于CAS和volatile关键字

2 locks包

1 Lock接口

• Lock接口概述
  • 锁是用来控制多个线程访问共享资源的方式，一个锁能够防止多个线程同时访问共享资源
  • 在Lock接口出现之前，Java程序主要是靠synchronized关键字实现锁功能的

而JDK5之后JUC包中增加了Lock接口，Lock接口提供了与synchronized一样的锁功能

• Lock接口是JUC包提供的一种全新的互斥同步手段
• Lock接口的实现类通过CAS操作和volatile关键字来保证线程安全

• 通常使用Lock的方式如下

  Lock lock = new ReentrantLock();    //ReentrantLock是Lock的实现类
  lock.lock();
  try {
    .......
  } finally {
    lock.unlock();
  }

  • synchronized同步块执行完成或者遇到异常时锁会自动释放，而Lock必须调用**unlock()**方法释放锁

因此为了避免同步代码出现异常影响解锁，需要在finally块中释放锁

• 虽然Lock接口失去了像synchronize关键字隐式加锁解锁的便捷性，但是却拥有了锁获取和释放的可操作性、可中断的获取锁以及超时获取锁等多种synchronized关键字所不具备的同步特性

• Lock接口API

Lock接口定义了6个方法

//获取锁
void lock(); 
//获取锁的过程能够响应中断
void lockInterruptibly() throws InterruptedException;
//非阻塞式响应中断能立即返回，获取锁返回true反之返回fasle
boolean tryLock();
//超时获取锁，在超时内或者未中断的情况下能够获取锁
boolean tryLock(long time, TimeUnit unit) throws InterruptedException;
//释放锁
unlock();
//创建一个与Lock对象绑定的条件对象，当前线程必须获得了锁才能在条件对象上等待
//线程等待时会先释放锁，当再次获取锁时才能从等待中返回
Condition newCondition();

• Lock的实现
  • ReentrantLock是常用的Lock接口实现类
  • ReentrantLock的源码并没有多少代码，并且ReentrantLock基本所有方法的实现都是调用了其静态内部类Sync（同步器）中的方法，而Sync类继承自抽象类AbstractQueuedSynchronizer（AQS）。
  • 因此要想理解Lock的实现原理关键是要理解AQS

2 AQS同步器

1 简介

• AQS概述
  • AQS全称AbstractQueuedSynchronizer（抽象阻塞同步器），是位于locks包下的抽象类，是一种通用的同步器机制，JUC中的同步器都继承自该类
  • 同步器是用来构建阻塞式锁和其他同步组件的基础框架

在AQS同步器出现之前，程序员会自己通过一种同步器去实现另一种相近的同步器，这显然不够优雅，因此在JDK5引入AQS同步器

• AQS同步器自身没有实现任何同步接口（如Lock），它仅仅是定义了锁状态的若干获取和释放方法来供自定义同步组件的使用
• AQS同步器既支持独占式获取锁状态，也可以支持共享式获取锁状态（独占模式是只有一个线程能够获取锁，而共享模式可以允许多个线程获取锁），这样就可以方便的实现不同类型的同步组件
• 常用的Lock接口实现类ReentrantLock就是基于AQS的

• AQS同步器与锁的关系
  • 同步器是实现锁（也可以是任意同步组件）的关键，在锁的实现中聚合同步器，利用同步器实现锁的语义。
  • 锁面向使用者，其定义了使用者与锁交互的接口，隐藏了实现细节

同步器是面向锁的实现者，它简化了锁的实现方式、屏蔽了锁状态的管理、线程的排队、等待和唤醒等底层操作。

• 同步组件通过调用同步器提供的方法可以很方便地实现需要的功能

锁和同步器很好的隔离了使用者和实现者所需关注的领域

• AQS的实现特点
  • AQS的核心是锁的状态，用int类型属性state来表示
  • 继承AQS的子类需要重写AQS的几个方法来自定义维护state的方式，进而控制如何获取锁和释放锁

AQS提供了以下方法来维护state

  - `**getState()**` - 获取state
  - `**setState()**` - 设置state
  - `**compareAndSetState()**` - CAS操作设置state，注意**使用CAS设置state仅保证设置state时的原子性，并不会反复重试，AQS的原理仍然是阻塞式的锁**

• AQS提供了基于FIFO的阻塞队列，类似于Monitor的EntryList，不过Monitor的EntryList是基于C++实现的，而AQS的阻塞队列是基于Java实现的
• AQS提供了条件变量来实现等待、唤醒机制，支持多个条件变量，类似于Monitor的WaitSet，不过AQS支持多个条件变量，而Monitor仅隐式支持一个

• AQS的使用

AQS被推荐定义为自定义同步组件中继承自AQS的静态内部类，例如ReentrantLock的定义

public class ReentrantLock implements Lock, java.io.Serializable {
    private final Sync sync;
    //Sync是抽象的是因为ReentrantLock内部还实现了FairSync和NonfairSync，它们都继承自Sync
    abstract static class Sync extends AbstractQueuedSynchronizer {
        ...
    }
    ...
}

2 AQS的设计模式

• 模板方法设计模式

AQS使用模板方法设计模式，即AQS将一些方法开放给子类进行重写，而AQS给同步组件所提供模板方法又会重新调用其被子类所重写的方法

• AQS提供的模板方法

在同步组件的实现中，AQS是核心部分，同步组件的实现者通过使用AQS提供的模板方法实现同步组件语义
模板方法不需要重写，但是模板方法会调用被子类重写的方法

• **void acquire(int arg)**

独占式获取同步状态，如果获取失败则将线程插入阻塞队列进行等待

• **void acquireInterruptibly(int arg)**

与acquire()方法相同，但在阻塞队列中等待的时候可以被中断

• **boolean tryAcquireNanos(int arg, long nanosTimeout)**

在**acquireInterruptibly()**基础上增加了超时等待功能，在超时时间内没有获得同步状态返回false

• **boolean release(int arg)**

释放同步状态，该方法会唤醒在阻塞队列中的下一个节点

• AQS可重写的方法（独占模式）

AQS中被protected修饰的方法可以根据需要被继承自AQS的子类重写，只需要重写以下三个方法

• **protected boolean tryAcquire(int arg)**

独占式尝试使用CAS获取一次锁（实际上是尝试使用CAS设置锁状态），获取失败时返回false，否则返回true
实现该方法需要查询当前锁状态并判断锁状态是否符合预期，然后再通过CAS操作设置锁状态

• **protected boolean tryRelease()**

独占式尝试释放锁

• protected boolean isHeldExclusively()

当前锁是否在独占模式下被线程占用，即表示锁是否被当前线程所独占

3 自定义独占式不可重入锁

• 实现Lock接口需要实现多个方法，如果方法内部的逻辑全部自己实现比较困难。但是我们自定义的同步器已经实现了大部分锁的底层逻辑，因此我们实现Lock接口的方法时只需要编写很少一部分代码

  class MyLock implements Lock {
    MySync sync = new MySync();
    //自定义非重入同步器
    static class MySync extends AbstractQueuedSynchronizer {
        //尝试获取锁
        @Override
        protected boolean tryAcquire(int acquires) {
            //尝试使用CAS操作将state设置为1
            if (compareAndSetState(0, 1)) {
                //加锁成功，将锁的owner设置为当前线程
                setExclusiveOwnerThread(Thread.currentThread());
                return true;
            }
            return false;
        }
        //尝试释放锁
        @Override
        protected boolean tryRelease(int acquires) {
            //注意以下两语句的顺序不能颠倒
            //因为AQS的state被volatile修饰，而volatile的写屏障可以保证在此之前语句的操作被其他线程看到
            setExclusiveOwnerThread(null);
            setState(0);
            return true;
        }
        //是否持有独占锁
        @Override
        protected boolean isHeldExclusively() {
            return getState() == 1;
        }
        protected Condition newCondition() {
            return new ConditionObject();
        }
    }
    @Override
    //尝试加锁，若不成功则进入阻塞队列
    public void lock() {
        sync.acquire(1);
    }
    @Override
    //尝试加锁，若不成功则进入阻塞队列，可被打断
    public void lockInterruptibly() throws InterruptedException {
        sync.acquireInterruptibly(1);
    }
    @Override
    //尝试加锁一次，若不成功则返回，不进入阻塞队列
    public boolean tryLock() {
        return sync.tryAcquire(1);
    }
    @Override
    //尝试加锁，若不成功则进入阻塞队列，阻塞时间有上限
    public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
        return sync.tryAcquireNanos(1, unit.toNanos(time));
    }
    @Override
    //释放锁
    public void unlock() {
        sync.release(1);
    }
    @Override
    //生成条件变量
    public Condition newCondition() {
        return sync.newCondition();
    }
  }

• 测试 ```java MyLock lock = new MyLock(); new Thread(() -> { log.debug(“locking…”); lock.lock(); try {

    log.debug("get lock!");
    Thread.sleep(5000);

  } catch (InterruptedException e) {

    e.printStackTrace();

  } finally {

    log.debug("unlocking...");
    lock.unlock();

  } }, “t1”).start();

new Thread(() -> { log.debug(“locking…”); lock.lock(); try { log.debug(“get lock!”); } finally { log.debug(“unlocking…”); lock.unlock(); } }, “t2”).start();

- 上述代码输出
```java
15:24:38 [t1] c.Test - locking...
15:24:38 [t1] c.Test - get lock!
15:24:38 [t2] c.Test - locking...
15:24:43 [t1] c.Test - unlocking...
15:24:43 [t2] c.Test - get lock!
15:24:43 [t2] c.Test - unlocking...

4 AQS原理

• 同步组件依赖于AQS，同步组件的方法大量调用了AQS的模板方法（即AQS的API），因此了解了AQS的实现原理也就能很轻松的了解同步组件的实现原理

1 AQS的属性

//阻塞队列的头节点
private transient volatile Node head;
//阻塞队列的尾节点
private transient volatile Node tail;
//同步状态
private volatile int state;
//独占模式下的持有同步状态的线程
//该属性是AQS父类中的属性，放在此处方便阅读
private transient Thread exclusiveOwnerThread

2 阻塞队列

• 阻塞队列概述

  • 当同步状态被某个线程占有，其他请求该同步状态的线程将会阻塞，从而进入阻塞队列。
  • 就数据结构而言，队列的实现方式无外乎两者一是通过数组的形式，另外一种则是链表的形式。AQS中的同步队列则是通过链式方式进行实现。

• 阻塞队列节点的属性

在AQS有一个静态内部类Node，表示阻塞队列中的节点，其属性如下

//节点等待状态
volatile int waitStatus; 
//当前节点/线程的前驱节点
volatile Node prev; 
//当前节点/线程的后继节点
volatile Node next; 
//加入阻塞队列的线程引用
volatile Thread thread;

• 从上述属性我们可以看出阻塞队列是一个双向链表

• 节点的等待状态

  //节点从同步队列中取消
  static final int CANCELLED =  1;
  //后继节点的线程处于等待状态
  static final int SIGNAL    = -1;
  //当前节点位于条件变量的等待队列中
  static final int CONDITION = -2;
  //初始状态
  static final int INITIAL = 0;

• AQS与阻塞队列

  • AQS的属性中有两个重要属性

    //阻塞队列的头节点
    private transient volatile Node head;
    //阻塞队列的尾节点
    private transient volatile Node tail;

  • AQS通过阻塞队列的头节点和尾结点来管理阻塞队列，实现包括获取锁失败的线程进行入队，释放锁时对阻塞队列中的线程进行通知等核心方法

3 获取同步状态原理

获取同步状态方法源码

• **void acquire(int arg)**

  public final void acquire(int arg) {
    //如果尝试获取同步状态失败，则创建一个Node对象并将其加入到阻塞队列中
    if (!tryAcquire(arg) &&
        acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
  }

• **boolean acquireQueued(final Node node, int arg)**

  final boolean acquireQueued(final Node node, int arg) {
    boolean interrupted = false;
    try {
        for (;;) {
            //获取node的前驱节点
            final Node p = node.predecessor();
            //如果前驱节点是哨兵，则再次调用tryAcquire()尝试获取同步状态
            if (p == head && tryAcquire(arg)) {
                //成功获取同步状态，从阻塞队列中摘掉node
                setHead(node);
                p.next = null; // help GC
                return interrupted;
            }
            //获取同步状态再次失败，调用shouldParkAfterFailedAcquire()方法，阻塞当前线程
            if (shouldParkAfterFailedAcquire(p, node))
                interrupted |= parkAndCheckInterrupt();
        }
    } catch (Throwable t) {
        cancelAcquire(node);
        if (interrupted)
            selfInterrupt();
        throw t;
    }
  }

• **boolean shouldParkAfterFailedAcquire(Node pred, Node node)**

  private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
        int ws = pred.waitStatus;
        if (ws == Node.SIGNAL)
            //This node has already set status asking a release to signal it, 
            //so it can safely park.
            return true;
        if (ws > 0) {
            //Predecessor was cancelled. Skip over predecessors and indicate retry.
            do {
                node.prev = pred = pred.prev;
            } while (pred.waitStatus > 0);
            pred.next = node;
        } else {
            //waitStatus must be 0 or PROPAGATE.  
            //Indicate that we need a signal, but don't park yet.  
            //Caller will need to retry to make sure it cannot acquire before parking.
            pred.compareAndSetWaitStatus(ws, Node.SIGNAL);
        }
        return false;
    }

• **boolean parkAndCheckInterrupt()**

  private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this);
    return Thread.interrupted();
  }

获取同步状态流程分析
现假设已经线程0持有了同步状态，此时线程1尝试获取同步状态

1. 线程0持有同步状态，没有发生竞争

1. 线程1尝试获取同步状态失败（子类重写的tryAcquire()逻辑，假设逻辑是如果state已经是1则返回false），执行acquireQueued(addWaiter(Node.EXCLUSIVE), arg))方法
2. 进入addWaiter()方法，该方法用于加入节点至阻塞队列

• 图中三角表示Node的waitStatus，0为默认正常状态
• 第一个Node称为Dummy或哨兵或头节点（惰性创建），用来占位，并不关联线程

头节点虽然不关联线程（指向null），但是头节点变量本身并不为null，而是指向一个node对象

1. 进入acquireQueued()方法

线程1会在acquireQueued()方法中再次尝试获取同步状态

• 线程1在一个死循环中不断尝试获得锁，如果失败则调用LockSupport.park()进入等待状态
• 如果关联线程1的节点是紧邻哨兵的节点，那么再次尝试获取锁，此时线程0没有释放锁，线程1获取同步状态再次失败
• 进入shouldParkAfterFailedAcquire()方法，此时前驱节点的waitStatus为0，将前驱节点的waitStatus改为 -1，方法返回false
  • 节点的waitStatus为-1表示该节点有责任唤醒其后继节点
• 再次进入循环尝试获取同步状态，若再次失败，进入shouldParkAfterFailedAcquire()方法，此时该方法返回true，并进入**parkAndCheckInterrupt()**方法
  1. parkAndCheckInterrupt()方法调用LockSupport.park(this)阻塞当前线程（线程1）

4 释放同步状态原理

释放同步状态方法方法源码

• **boolean release(int arg)**

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

• **void unparkSuccessor(Node node)**

  private void unparkSuccessor(Node node) {
    int ws = node.waitStatus;
    //将头节点waitStatus置为0
    if (ws < 0)
        node.compareAndSetWaitStatus(ws, 0);
    //找到与头节点最近的waitStatus为-1的后继节点
    Node s = node.next;
    if (s == null || s.waitStatus > 0) {
        s = null;
        for (Node p = tail; p != node && p != null; p = p.prev)
            if (p.waitStatus <= 0)
                s = p;
    }
    //唤醒后继节点
    if (s != null)
        LockSupport.unpark(s.thread);
  }

释放同步状态方法流程分析
现假设已经线程0持有了同步状态，多个线程竞争失败

1. 初始状态，多个线程阻塞

1. 执行tryRelease()方法，线程0尝试释放锁，如果成功，则将exclusiveOwnerThread置为null、state置为0

tryRelease()方法会调用release()来唤醒一个线程

1. 执行release()方法，由于当前头节点不为null且头节点的waitStatus为-1，因此进入unparkSuccessor()方法：找到阻塞队列中离头节点最近的一个waitStatus为-1的Node（图中为表示线程1的Node），调用LockSupport.unpark()恢复其运行
2. 线程1被唤醒，继续执行其acquireQueued()方法，尝试获取同步状态

若获取成功，则：

• 将exclusiveOwnerThread设置为Thread-1、state设置为1
• 将表示线程1的Node置为头节点
• 将之前的头节点与阻塞队列断开，此后该头节点可被垃圾回收

若获取失败，如又有一个新的线程4获取到了同步状态，那么线程1获取同步状态失败，重新调用LockSupport.park()方法阻塞线程1

由于等待最久的线程1不一定能获取到同步状态，因此这是不公平的策略

5 可打断原理

• 如果希望在等待锁的时候可以被打断，则基于AQS的同步组件会调用**acquireInterruptibly()**而不是**acquire()**方法

• **void acquireInterruptibly(int arg) throws InterruptedException**

  public final void acquireInterruptibly(int arg) throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    //尝试获取同步状态失败后，进入doAcquireInterruptibly()而不是acquireQueued()
    if (!tryAcquire(arg))
        doAcquireInterruptibly(arg);
  }

• **void doAcquireInterruptibly(int arg)**

  private void doAcquireInterruptibly(int arg)
    throws InterruptedException {
    final Node node = addWaiter(Node.EXCLUSIVE);
    try {
        for (;;) {
            final Node p = node.predecessor();
            if (p == head && tryAcquire(arg)) {
                setHead(node);
                p.next = null; // help GC
                return;
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                //注意这里与acquireQueued()的区别
                //这里会直接抛出异常，而acquireQueued()不会
                throw new InterruptedException();
        }
    } catch (Throwable t) {
        cancelAcquire(node);
        throw t;
    }
  }

• **boolean parkAndCheckInterrupt()**

  private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this);
    // 当park()被打断时，会返回true并清除打断标记
    return Thread.interrupted();
  }

6 条件变量等待原理

• 条件变量概述

  • 每一个条件变量就对应着一个等待队列（类似于Monitor的WaitSet）
  • 条件变量的实现类是ConditionObject
  • ConditionObject是AQS的内部类

• 等待与唤醒

  • 等待

可以在条件变量上调用await()方法，使得当前线程在该条件上等待，进入WAITING状态

• 唤醒

可以在条件变量上调用signal()或signalAll()方法，唤醒条件变量上的一个或全部线程

• 条件变量的属性

  //等待队列的头节点
  private transient Node firstWaiter;
  //等待队列的尾结点
  private transient Node lastWaiter;

条件变量等待方法源码

• **void await() throws InterruptedException**

  public final void await() throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    //将代表当前线程的Node加入当前条件的等待队列中
    //等待队列是一个链表，类似于同步器的阻塞队列
    Node node = addConditionWaiter();
    //将节点对应的持有的锁全部释放掉并唤醒阻塞队列中的线程
    int savedState = fullyRelease(node);
    int interruptMode = 0;
    while (!isOnSyncQueue(node)) {
        //进入WAITING状态，等待被唤醒
        LockSupport.park(this);
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
    }
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;
    if (node.nextWaiter != null) // clean up if cancelled
        unlinkCancelledWaiters();
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);
  }

• **Node addConditionWaiter()**

  private Node addConditionWaiter() {
    //###如果当前线程没有持有锁，则抛出异常###
    //因此await()与Object::wait()方法一样，调用的前提是持有锁
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    Node t = lastWaiter;
    // If lastWaiter is cancelled, clean out.
    if (t != null && t.waitStatus != Node.CONDITION) {
        unlinkCancelledWaiters();
        t = lastWaiter;
    }
    //新建一个代表当前线程的节点并加入到当前条件的等待队列中
    Node node = new Node(Node.CONDITION);
    if (t == null)
        firstWaiter = node;
    else
        t.nextWaiter = node;
    lastWaiter = node;
    return node;
  }

条件变量等待流程分析

1. 初始状态，线程0获取锁，多个线程阻塞

1. 线程0在一个Condition对象上调用**await()**方法

await()方法会创建一个代表当前线程的Node对象（该Node对象的WaitStatus为-2，表示在Condition的等待队列中等待），之后将该Node加入Condition对象的等待队列中

• 该等待队列类似于AQS的阻塞队列，但是没有头节点

1. 线程0进入等待队列后，将调用fullRelease()，释放掉线程0的所有锁，并唤醒阻塞队列中的线程

7 条件变量唤醒原理

条件变量唤醒方法源码

• **void signal()**

  public final void signal() {
    //如果不是锁的持有者调用该方法则抛出异常
    if (!isHeldExclusively())
        throw new IllegalMonitorStateException();
    //唤醒Condition等待队列中的第一个线程
    Node first = firstWaiter;
    if (first != null)
        doSignal(first);
  }

• **void doSignal(Node first)**

  private void doSignal(Node first) {
    do {
        //将first节点从等待队列中断开
        if ( (firstWaiter = first.nextWaiter) == null)
            lastWaiter = null;
        first.nextWaiter = null;
      //将first节点转移到AQS的阻塞队列中
      //如果转移失败，则尝试唤醒first的后一个节点
    } while (!transferForSignal(first) &&
             (first = firstWaiter) != null);
  }

• **boolean transferForSignal(Node node)**

  final boolean transferForSignal(Node node) {
    //先将Node的WaitStatus从2转为0
    if (!node.compareAndSetWaitStatus(Node.CONDITION, 0))
        return false;
    //enq()方法将node加入到AQS的阻塞队列，如果成功了返回node的前驱节点
    Node p = enq(node);
    //将前驱节点的WaitStatue改为-1
    int ws = p.waitStatus;
    if (ws > 0 || !p.compareAndSetWaitStatus(ws, Node.SIGNAL))
        LockSupport.unpark(node.thread);
    return true;
  }

条件变量唤醒流程分析

1. 初始状态，Thread0在等待队列，Thread1持有锁

1. 持有锁的线程1调用Condition对象的signal()方法唤醒该对象等待队列的一个线程

singal()方法调用doSignal()方法将表示线程0的Node从Condition对象的等待队列中断开

1. doSignal()方法调用transferForSignal()方法，将代表线程0的Node的状态改为0，并将其插入到AQS阻塞队列的队尾，同时将Node的前驱节点的状态改为-1

• 需要注意的是，线程0在整个唤醒的过程中都处于**park()**状态（WAITING状态）没有改变，唤醒仅是让线程0可以参与后续锁的竞争

8 阻塞队列（同步队列）与等待队列的关系

5 Condition接口与Object监视器方法

• Condition接口与Object监视器方法概述
  • 任何一个Java对象都天然继承于Object类，在线程间实现通信的往往会应用到Object的几个方法，比如wait()、wait(long timeout)与notify()、notifyAll()几个方法实现等待/唤醒机制

同样的， 在JUC包的Lock体系下依然会有同样的方法实现等待/唤醒机制。

• Object的wait()和notify()/notifyAll()是与Monitor配合完成线程间的等待/唤醒机制

Condition与AQS配合（ConditionObject是AQS的内部类）完成等待通知机制
前者是Java底层级别的，后者是语言级别的，具有更高的可控制性和扩展性。

• Condition接口与Object监视器方法的对比 | 对比项 | Object Monitor | Condition | | —- | —- | —- | | 前置条件 | 线程获取到锁对象 |
  - 线程调用lock.lock()获取到锁
  - 存在Condition对象
  | | 调用方式 | 锁对象上调用 | Condition对象上调用 | | 等待队列（WaitSet）个数 | 1 | 多个 | | 线程进入等待队列时释放锁 | √ | √ | | 线程进入等待队列时响应中断 | √ | √ | | 线程进入等待队列时不响应中断 | ×（必须响应中断） | √ | | 线程进入等待队列时进入超时等待状态 | √ | √ | | 唤醒等待队列中的一个线程 | √ | √ | | 唤醒等待队列中的全部线程 | √ | √ |

3 ReentrantLock类

1 ReentrantLock简介

• ReentrantLock（重入锁）类概述

  • ReentrantLock是java.util.concurrent.locks包下的类
  • ReentrantLock是Lock接口的一个实现类，也是在实际编程中使用频率很高的一个锁

• ReentrantLock基本语法

  // 获取锁
  reentrantLock.lock();
  try {
    // 临界区
  } finally {
    // 释放锁
    reentrantLock.unlock();
  }

  • synchronized是在Java语法级别（关键字级别）保护临界区，而ReentrantLock则是在对象级别保护临界区，因此在使用ReentrantLock前需要先创建ReentrantLock对象
  • Lock应该确保在finally块中释放锁，否则一旦临界区中的代码抛出异常时，则可能永远不会释放持有的锁

2 ReentrantLock与synchronized

• ReentrantLock与synchronized功能的区别

ReentrantLock与synchronized一样是可重入的，在基本用法上ReentrantLock也与synchronized很相似，只是代码写法上稍有区别。不过ReentrantLock与synchronized相比增加了一些高级功能，主要是以下四项（a、b实际上是LockSupport的特性）

1. 超时加锁（主动避免长期等待锁释放）

当持有锁的线程长期不释放锁的时候，正在阻塞等待锁的线程可以选择放弃等待，改为处理其他事情

1. 可中断加锁（被动避免长期等待锁释放）

等待持有锁的线程释放锁的线程，也可以被中断
可中断可以避免死锁

1. 公平锁

• 公平锁是指多个线程在等待同一个锁时，必须按照申请锁的时间顺序来依次获得锁
• 非公平锁是指在锁被释放时，任何一个等待锁的线程都有机会获得锁
• synchronized中的锁是非公平的，ReentrantLock在默认情况下也是非公平的，但可以通过带布尔值的构造函数将ReentrantLock配置为公平锁
• ReentrantLock一旦使用公平锁，其性能将急剧下降，会明显影响吞吐量

1. 锁绑定多个条件

• 一个ReentrantLock对象可以同时绑定多个Condition对象（Condition类位于java.util.concurrent.locks包）
• 在synchronized中，锁对象的wait()和notify()/notifyAll()方法相配合，可以实现一个隐含的条件，但是无法和多于一个的条件相关联

ReentrantLock则可以通过newCondition()方法与多个条件相关联

• 换句话说，synchronized只有一个WaitSet，而ReentrantLock可以有多个WaitSet，每个WaitSet对应一个条件，线程可以根据不同的条件进入不同的WaitSet，这样可以做到更为精准的唤醒，减少虚假唤醒

• ReentrantLock与synchronized的选择
  • 由于JDK6针对synchronized关键字的优化，两者在性能上已相差无几，因此不再是选择的参考因素
  • 在功能上，ReentrantLock是synchronized的超集，如果需要使用ReentrantLock独有的功能肯定要选择ReentrantLock
  • 在synchronized和ReentrantLock都可满足需要时，优先选择synchronized，原因如下
    1. synchronized是在Java语法层面上的同步，足够清晰简单，被更多程序员熟知
    2. Lock必须确保在finally块中释放锁，但这是由程序员自己来保证的，而使用synchronized的话则可以由JVM来确保即使出现异常，锁也能被自动释放
    3. 从长远来看，synchronized更容易被继续优化，而ReentrantLock则难以继续优化

3 ReentrantLock的使用

1. 可中断加锁

注意如果想让进入BLOCKED状态的线程可被中断，那么在加锁时就不能使用reentrantLock.lock()方法，而必须使用**reentrantLock.lockInterruptibly()**方法

private static ReentrantLock reentrantLock = new ReentrantLock();
public static void main(String[] args) {
    Thread t1 = new Thread(() -> {
        log.debug("启动...");
        try {
            //使用可被打断的方式加锁
            reentrantLock.lockInterruptibly();
        } catch (InterruptedException e) {
            e.printStackTrace();
            log.debug("等锁的过程中被打断");
            return;
        }
        try {
            log.debug("获得了锁");
        } finally {
            reentrantLock.unlock();
        }
    }, "t1");
    reentrantLock.lock();
    try {
        log.debug("主线程获得了锁");
        t1.start();
        Thread.sleep(1000);
        t1.interrupt();
        log.debug("执行打断");
    } catch (InterruptedException e) {
        e.printStackTrace();
    } finally {
        reentrantLock.unlock();
    }
}

• 上述代码输出

  15:48:49 [main] c.Test - 主线程获得了锁
  15:48:49 [t1] c.Test - 启动...
  15:48:50 [main] c.Test - 执行打断
  15:48:50 [t1] c.Test - 等锁的过程中被打断
  java.lang.InterruptedException
  at java.base/java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireInterruptibly(AbstractQueuedSynchronizer.java:944)
  at java.base/java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireInterruptibly(AbstractQueuedSynchronizer.java:1263)
  at java.base/java.util.concurrent.locks.ReentrantLock.lockInterruptibly(ReentrantLock.java:317)
  at Test.lambda$main$0(Test.java:18)
  at java.base/java.lang.Thread.run(Thread.java:834)

1. 超时等待

超时等待需要使用**reentrantLock.tryLock(long timeout)**方法实现

• 如果该方法不传入参数，则线程在尝试获取锁失败后将直接退出阻塞状态
• 如果传入参数则会在阻塞时间到达timeout后退出阻塞 ```java private static final ReentrantLock reentrantLock = new ReentrantLock();

public static void main(String[] args) { Thread t1 = new Thread(() -> { log.debug(“启动…”); if (!reentrantLock.tryLock()) { log.debug(“获取立刻失败，返回”); return; } try { log.debug(“获得了锁”); } finally { reentrantLock.unlock(); } }, “t1”);

reentrantLock.lock();
try {
    log.debug("主线程获得了锁");
    t1.start();
    Thread.sleep(2000);
} catch (InterruptedException e) {
    e.printStackTrace();
} finally {
    reentrantLock.unlock();
}

}

   - 上述代码输出
```java
16:00:10 [main] c.Test - 主线程获得了锁
16:00:10 [t1] c.Test - 启动...
16:00:10 [t1] c.Test - 获取立刻失败，返回

1. 公平锁

如果想将ReentrantLock配置为公平锁，实例化ReentrantLock对象时，给ReentrantLock构造函数传入true即可

• ReentrantLock构造函数如下

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

1. 条件变量

• 使用**reentrantLock.newCondition()**可以产生一个新的条件对象Condition
• 等待

在Condition对象上调用**condition.await()**方法即可在指定条件上等待

• 在执行condition.await()方法之前必须先获得锁，执行该方法后将释放锁
• 执行await()方法的线程可以被唤醒也可以被中断（类似wait()），中断将抛出异常
  • 唤醒

在Condition对象上调用**condition.signal()**/**condition.signalAll()**方法即可唤醒某个或所有等待指定条件的线程

static ReentrantLock lock = new ReentrantLock();
static Condition waitCigaretteQueue = lock.newCondition();
static Condition waitBreakfastQueue = lock.newCondition();
static volatile boolean hasCigarette = false;
static volatile boolean hasBreakfast = false;
public static void main(String[] args) throws InterruptedException {
    new Thread(() -> {
        try {
            lock.lock();
            while (!hasCigarette) {
                try {
                    waitCigaretteQueue.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            log.debug("等到了它的烟");
        } finally {
            lock.unlock();
        }
    }).start();
    new Thread(() -> {
        try {
            lock.lock();
            while (!hasBreakfast) {
                try {
                    waitBreakfastQueue.await();
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
            log.debug("等到了它的早餐");
        } finally {
            lock.unlock();
        }
    }).start();
    Thread.sleep(1000);
    //唤醒等待早餐的线程
    sendBreakfast();
    Thread.sleep(1000);
    //唤醒等待香烟的线程
    sendCigarette();
}
private static void sendCigarette() {
    lock.lock();
    try {
        log.debug("送烟来了");
        hasCigarette = true;
        //随机唤醒一个waitCigaretteQueue条件上的线程
        waitCigaretteQueue.signal();
    } finally {
        lock.unlock();
    }
}
private static void sendBreakfast() {
    lock.lock();
    try {
        log.debug("送早餐来了");
        hasBreakfast = true;
        //唤醒所有waitCigaretteQueue条件上的线程
        waitBreakfastQueue.signalAll();
    } finally {
        lock.unlock();
    }
}

• 上述代码输出

  16:38:24 [main] c.Test - 送早餐来了
  16:38:24 [Thread-1] c.Test - 等到了它的早餐
  16:38:25 [main] c.Test - 送烟来了
  16:38:25 [Thread-0] c.Test - 等到了它的烟

4 ReentrantLock原理

• ReentrantLock类结构图

• ReentrantLock内部实现了一个抽象类Sync，其继承自AQS
• FairSync和NonfairSync均继承自Sync，分别实现了公平同步器和非公平同步器’，ReentrantLock实际使用的是这两个同步器
• FairSync和NonfairSync的区别是对于**tryAcquire()**方法实现的不同，其他方法都继承自Sync

1 ReentrantLock源码

public class ReentrantLock implements Lock, java.io.Serializable {
    private final Sync sync;
    abstract static class Sync extends AbstractQueuedSynchronizer{...}
    //非公平同步器
    static final class NonfairSync extends Sync{...}
    //公平同步器
    static final class FairSync extends Sync{...}
    //ReentrantLock默认为非公平锁
    public ReentrantLock() {
        sync = new NonfairSync();
    }
    //可以指定ReentrantLock为公平锁
    public ReentrantLock(boolean fair) {
        sync = fair ? new FairSync() : new NonfairSync();
    }
    //直接调用AQS的acquire方法
    public void lock() {
        sync.acquire(1);
    }
    public void lockInterruptibly() throws InterruptedException {
        sync.acquireInterruptibly(1);
    }
    public boolean tryLock() {
        return sync.nonfairTryAcquire(1);
    }
    public boolean tryLock(long timeout, TimeUnit unit)
            throws InterruptedException {
        return sync.tryAcquireNanos(1, unit.toNanos(timeout));
    }
    public void unlock() {
        sync.release(1);
    }
    public Condition newCondition() {
        return sync.newCondition();
    }
    ...
}

• 从上述代码中可以看到ReentrantLock的核心方法均是调用的非公平同步器或公平同步器的方法

2 NonfairSync源码

static final class NonfairSync extends Sync {
    protected final boolean tryAcquire(int acquires) {
        return nonfairTryAcquire(acquires);
    }
}

• NonfairSync的方法与Sync的方法基本一致，只是起到了一个包装的作用

3 FairSync源码

static final class FairSync extends Sync {
    //公平版本的tryAcquire
    @ReservedStackAccess
    protected final boolean tryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
        int c = getState();
        if (c == 0) {
            //注意这里与nonfairTryAcquire()方法的区别
            //当锁可用时，公平同步器会先检查AQS阻塞队列中是否有节点
            //如果AQS没有节点则去竞争锁，否则不参与竞争，直接进入阻塞队列
            if (!hasQueuedPredecessors() &&
                compareAndSetState(0, acquires)) {
                setExclusiveOwnerThread(current);
                return true;
            }
        }
        else if (current == getExclusiveOwnerThread()) {
            int nextc = c + acquires;
            if (nextc < 0)
                throw new Error("Maximum lock count exceeded");
            setState(nextc);
            return true;
        }
        return false;
    }
}

3 Sync源码

abstract static class Sync extends AbstractQueuedSynchronizer {
    //tryAcquire()方法在Sync的两个子类中实现，NonefairSync会调用该方法
    //该方法表现为不公平策略
    @ReservedStackAccess
    final boolean nonfairTryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
        int c = getState();
        //如果当前锁无人获得，则将state设置为1
        if (c == 0) {
            if (compareAndSetState(0, acquires)) {
                setExclusiveOwnerThread(current);
                return true;
            }
        }
        //如果当前线程已经获得了锁，则返回true，即发生了锁重入
        else if (current == getExclusiveOwnerThread()) {
            //锁重入后令state++
            int nextc = c + acquires;
            if (nextc < 0) // overflow
                throw new Error("Maximum lock count exceeded");
            setState(nextc);
            return true;
        }
        return false;
    }
    @ReservedStackAccess
    protected final boolean tryRelease(int releases) {
        //锁释放时，令state--
        int c = getState() - releases;
        if (Thread.currentThread() != getExclusiveOwnerThread())
            throw new IllegalMonitorStateException();
        boolean free = false;
        //当state为0时才释放锁，因为此时当前线程还持有锁
        if (c == 0) {
            free = true;
            setExclusiveOwnerThread(null);
        }
        setState(c);
        //state大于0，此时并不释放锁，只是将锁重入的计数（state）减1
        return free;
    }
    protected final boolean isHeldExclusively() {
        // While we must in general read state before owner,
        // we don't need to do so to check if current thread is owner
        return getExclusiveOwnerThread() == Thread.currentThread();
    }
    final ConditionObject newCondition() {
        return new ConditionObject();
    }
    final Thread getOwner() {
        return getState() == 0 ? null : getExclusiveOwnerThread();
    }
    final int getHoldCount() {
        return isHeldExclusively() ? getState() : 0;
    }
    final boolean isLocked() {
        return getState() != 0;
    }
    private void readObject(java.io.ObjectInputStream s)
        throws java.io.IOException, ClassNotFoundException {
        s.defaultReadObject();
        setState(0); // reset to unlocked state
    }
}

4 锁重入原理

• **Sync::nonfairTryAcquire()**和**Sync::boolean tryRelease()** ```java @ReservedStackAccess final boolean nonfairTryAcquire(int acquires) { final Thread current = Thread.currentThread(); int c = getState();

  //如果当前锁无人获得，则将state设置为1 if (c == 0) {

    if (compareAndSetState(0, acquires)) {
        setExclusiveOwnerThread(current);
        return true;
    }

  }

  //如果当前线程已经获得了锁，则返回true表示加锁成功，此时发生了锁重入 else if (current == getExclusiveOwnerThread()) {

    //锁重入后令state++
    int nextc = c + acquires;
    if (nextc < 0) // overflow
        throw new Error("Maximum lock count exceeded");
    setState(nextc);
    return true;

  } return false; }

@ReservedStackAccess protected final boolean tryRelease(int releases) { //锁释放时，令state— int c = getState() - releases; if (Thread.currentThread() != getExclusiveOwnerThread()) throw new IllegalMonitorStateException(); boolean free = false;

//当state为0时才释放锁，因为此时当前线程还持有锁
if (c == 0) {
    free = true;
    setExclusiveOwnerThread(null);
}
setState(c);
//state大于0，此时并不释放锁，只是将锁重入的计数（state）减1
return free;

}

   - 从上述代码中可以看到，**在加锁和释放锁时通过state的值和exclusiveOwnerThread属性来判断**
      - 当获取锁时，若state的值大于0，则说明锁已经被线程获取，又判断持有锁的线程和当前线程是否相等，若相等，则**加锁成功，state的值加1**
      - 当**释放锁时，state的值减1**，**若此时state的值为0，则释放锁成功并返回true**，否则返回false
<a name="O0fAk"></a>
#### 4 非公平锁原理
- **当ReentrantLock初始化为非公平锁时，使用的同步器是NonfairSync，此时ReentrantLock的属性Sync指向NonfairSync对象**
<br />
- `**ReentrantLock::lock()**`
```java
public void lock() {
    sync.acquire(1);
}

• 可以看到调用lock时实际调用的是AQS::acquire()方法，而acquire()方法又会调用NonfairSync::tryAcquire()方法，而tryAcquire()方法直接调用Sync::nonfairTryAcquire()

• **Sync::nonfairTryAcquire()**

  @ReservedStackAccess
  final boolean nonfairTryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();
    //如果当前锁无人获得，则将state设置为1
    //由于这里直接尝试使用CAS操作去竞争锁，不检查AQS阻塞队列，因此是非公平的策略
    if (c == 0) {
        if (compareAndSetState(0, acquires)) {
            setExclusiveOwnerThread(current);
            return true;
        }
    }
    //如果当前线程已经获得了锁，则返回true，即发生了锁重入
    else if (current == getExclusiveOwnerThread()) {
        //锁重入后令state++
        int nextc = c + acquires;
        if (nextc < 0) // overflow
            throw new Error("Maximum lock count exceeded");
        setState(nextc);
        return true;
    }
    return false;
  }

  • 从上述代码可以看到，任何线程在调用**nonfairTryAcquire()**方法时都会直接去竞争锁，而不会去检查AQS阻塞队列，这可能导致AQS阻塞队列的线程获取不到锁，最终导致不公平

5 公平锁原理

• 当ReentrantLock初始化为公平锁时，使用的同步器是FairSync，此时ReentrantLock的属性Sync指向FairSync对象

• FairSync与NonfairSync的唯一区别就是**tryAcquire()**方法的区别
  • 对于FairSync来说，新的线程尝试获取锁时，如果发现AQS阻塞队列中已经有线程在等待，则直接放弃竞争，进入阻塞队列
  • 具体实现细节见FairSync源码

4 ReentrantReadWriteLock类

1 简介

• ReentrantReadWriteLock（读写锁）概述
  • 在并发场景中用于解决线程安全的问题，我们几乎会高频率的使用到独占式锁，通常使用Java提供的关键字synchronized或者JUC包中实现了Lock接口的ReentrantLock。

上述两种锁都是独占式获取锁，也就是在同一时刻只有一个线程能够获取锁

• 在一些大部分只是读数据而写数据很少的业务场景中，如果仅仅是读数据的话并不会影响数据正确性（不会出现脏读），而在这种业务场景下依然使用独占锁的话，很显然会影响性能。
• 针对这种读多写少的情况，JUC包还提供了另外一个实现Lock接口的ReentrantReadWriteLock类。

• ReentrantReadWriteLock特点
  • 读写锁允许同一时刻被多个读线程访问，但是在写线程访问时所有的读线程和其他的写线程都会被阻塞。
  • 读写锁中有两种锁，分别是读锁（ReadLock）和写锁（WriteLock），读锁可以用来保护共享数据的读取，写锁用来保护共享数据的写入，读锁与写锁相互配合可以保证
    • 读-读操作并行
    • 读-写操作串行
    • 写-写操作串行
  • 读写锁的其他特性如下
    • 公平锁

与ReentrantLock一样，读写锁默认非公平锁，但是也可以指定为公平锁

  - **重入性**

与ReentrantLock一样，支持锁的重入

  - **条件变量**

读锁不支持条件变量，写锁支持条件变量

  - **锁重入时不支持锁升级**

一个线程获取读锁后不能再获取写锁，必须先释放读锁再获取写锁

  - **锁重入时支持锁降级**

一个线程获取写锁后，还可以获取读锁
由于在写锁释放前可以获取读锁，因此线程获取读锁后再释放写锁，就完成了写锁到读锁的降级

2 ReentrantReadWriteLock使用实例

• 现定义一个线程安全的数据容器类DataContainer，其内部使用ReentrantReadWriteLock的读锁保护读取数据的read()方法，使用写锁来保护写入数据的write()方法

• DataContainer源码

  @Slf4j(topic = "c.DataContainer")
  class DataContainer {
    private Object data;
    //读写锁
    private final ReentrantReadWriteLock rw = new ReentrantReadWriteLock();
    //读锁
    private final ReentrantReadWriteLock.ReadLock rLock = rw.readLock();
    //写锁
    private final ReentrantReadWriteLock.WriteLock wLock = rw.writeLock();
    //读取操作使用读锁保护
    public Object read() {
        log.debug("获取读锁...");
        rLock.lock();
        try {
            log.debug("读取");
            Thread.sleep(1000);
            return data;
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            log.debug("释放读锁...");
            rLock.unlock();
        }
        return null;
    }
    //写入操作使用写锁保护
    public void write() {
        log.debug("获取写锁...");
        wLock.lock();
        try {
            log.debug("写入");
            Thread.sleep(1000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            log.debug("释放写锁...");
            wLock.unlock();
        }
    }
  }

• 测试

  @Slf4j(topic = "c.Test")
  public class Test {
    static DataContainer dataContainer = new DataContainer();
    public static void main(String[] args) throws InterruptedException {
        log.debug("读-读并发测试");
        readRead();
        Thread.sleep(3000); //等待读-读测试完毕
        log.debug("读-写并发测试");
        readWrite();
    }
    //读-读测试
    static void readRead() {
        new Thread(() -> {
            dataContainer.read();
        }, "r_t1").start();
        new Thread(() -> {
            dataContainer.read();
        }, "r_t2").start();
    }
    //读-写测试
    static void readWrite() throws InterruptedException {
        new Thread(() -> {
            dataContainer.read();
        }, "r_t3").start();
        Thread.sleep(100);  //等待一下，保证t1获取到读锁
        new Thread(() -> {
            dataContainer.write();
        }, "w_t1").start();
    }
  }

  测试结果

  16:53:11 [main] c.Test - 读-读并发测试
  16:53:11 [r_t1] c.DataContainer - 获取读锁...
  16:53:11 [r_t2] c.DataContainer - 获取读锁...
  16:53:11 [r_t1] c.DataContainer - 读取
  16:53:11 [r_t2] c.DataContainer - 读取
  16:53:12 [r_t2] c.DataContainer - 释放读锁...
  16:53:12 [r_t1] c.DataContainer - 释放读锁...
  16:53:14 [main] c.Test - 读-写并发测试
  16:53:14 [r_t3] c.DataContainer - 获取读锁...
  16:53:14 [r_t3] c.DataContainer - 读取
  16:53:14 [w_t1] c.DataContainer - 获取写锁...
  16:53:15 [r_t3] c.DataContainer - 释放读锁...
  16:53:15 [w_t1] c.DataContainer - 写入
  16:53:16 [w_t1] c.DataContainer - 释放写锁...

  • 从测试结果可以看到
    • 读-读操作可以并行
    • 读-写操作时，由于已经有线程开始了读操作，执行写操作的线程被阻塞

3 应用 - 实现Redis缓存与数据库一致性

• 应用实现的功能
  1. 查询数据

查询数据时先看Redis中是否有数据，有的话直接从Redis中查，没有的话则从MySQL中查并将结果存入Redis

1. 更新数据

更新数据时需要将数据更新到MySQL并将Redis中的缓存删除

1 非线程安全实现

• 源码

  @Service("userService")
  public class UserServiceImpl implements UserService {
    @Autowired
    UserMapper userMapper;
    @Autowired
    JedisPool jedisPool;
    @Override
    public User findOne(int id) {
        Jedis jedis = jedisPool.getResource();
        String key = "person:" + id;
        //查询数据时先从Redis缓存中查询
        User value = (User) SerializeUtil.unserialize(jedis.get(key.getBytes()));
        if (value != null) {
            System.out.print("Get from Redis: ");
            return value;
        }
        //Redis中没有查询到，从MySQL中查并将结果存入Redis
        value = userMapper.findByCondition(id);
        jedis.set(key.getBytes(), SerializeUtil.serialize(value));
        System.out.print("Get from MySQL: ");
        return value;
    }
    @Override
    public int update(User user) {
        Jedis jedis = jedisPool.getResource();
        String key = "person:" + user.getId();
        //先更新数据库再清理缓存
        int rows = userMapper.update(user);
        jedis.del(key);
        return rows;
    }
  }

• 测试 ```java ApplicationContext app = new ClassPathXmlApplicationContext(“applicationContext.xml”); UserService userService = (UserService) app.getBean(“userService”);

System.out.println(“==========> 查询”); System.out.println(userService.findOne(1)); System.out.println(userService.findOne(1)); System.out.println(userService.findOne(1));

System.out.println(“==========> 更新”); User user = new User(1, “常桐华”, “520”); System.out.println(userService.update(user)); System.out.println(userService.findOne(1));

   - 上述代码测试结果如下
```java
==========> 查询
14:47:59,308  INFO DruidDataSource:655 - {dataSource-1} inited
Get from MySQL: User{id=1, username='崔奕宸', password='123'}
Get from Redis: User{id=1, username='崔奕宸', password='123'}
Get from Redis: User{id=1, username='崔奕宸', password='123'}
==========> 更新
1
Get from MySQL: User{id=1, username='常桐华', password='520'}

上述代码中存在三个问题

1. 对于Redis的操作是非线程安全的
2. findOne()方法中，当多个线程同时首次查询同一个数据，多个线程可能都会到MySQL中去查询，导致缓存穿透
3. update()方法涉及两个重要操作：1. 删除缓存内容 2. 更新数据库，两个操作的顺序不同（即更新策略的不同）会导致不同的线程安全问题，分析如下

• 更新策略1：先删缓存再更新数据库

• 如上图所示的执行顺序，该更新策略可能导致某些线程一直查询到旧值而查询不到新值
  • 更新策略2：先更新数据库再删缓存

• 如上图所示的执行顺序，该更新策略可能导致少量的线程查询到旧值，但是该状况不会持续
• 对于不需要严格线程安全的场景，该更新策略比较适合

2 线程安全实现

• 对于上述出现的三个问题，可以通过ReentrantReadWriteLock来解决
  • 对MySQL或Redis的写操作使用写锁来保护
  • 对MySQL或Redis的读操作使用读锁来保护

源码如下

@Service("userService")
public class UserServiceImpl implements UserService {
    @Autowired
    UserMapper userMapper;
    @Autowired
    JedisPool jedisPool;
    @Autowired
    ReentrantReadWriteLock lock;
    @Override
    public User findOne(int id) {
        Jedis jedis = jedisPool.getResource();
        String key = "person:" + id;
        //需要读取Redis，加读锁保护
        lock.readLock().lock();
        User value;
        try {
            //查询数据时先从Redis缓存中查询
            value = (User) SerializeUtil.unserialize(jedis.get(key.getBytes()));
            if (value != null) {
                System.out.print("Get from Redis: ");
                return value;
            }
        } finally {
            lock.readLock().unlock();
        }
        lock.writeLock().lock();
        try {
            //可能多个线程试图查询数据库，因此这里需要双重检查防止缓存穿透
            value = (User) SerializeUtil.unserialize(jedis.get(key.getBytes()));
            if (value != null) {
                System.out.print("Get from Redis: ");
            } else {
                //Redis中没有查询到，从MySQL中查并将结果存入Redis
                value = userMapper.findByCondition(id);
                jedis.set(key.getBytes(), SerializeUtil.serialize(value));
                System.out.print("Get from MySQL: ");
            }
            return value;
        } finally {
            lock.writeLock().unlock();
        }
    }
    @Override
    public int update(User user) {
        Jedis jedis = jedisPool.getResource();
        String key = "person:" + user.getId();
        //需要写入数据库，加写锁保护
        lock.writeLock().lock();
        //先更新数据库再清理缓存
        try {
            int rows = userMapper.update(user);
            jedis.del(key);
            return rows;
        } finally {
            lock.writeLock().unlock();
        }
    }
}