Java 容器简介

容器简介

数组与容器

Java 中常用的存储容器就是数组和容器，二者有以下区别：

• 存储大小是否固定
  • 数组的长度固定；
  • 容器的长度可变。
• 数据类型
  • 数组可以存储基本数据类型，也可以存储引用数据类型；
  • 容器只能存储引用数据类型，基本数据类型的变量要转换成对应的包装类才能放入容器类中。

:bulb: 不了解什么是基本数据类型、引用数据类型、包装类这些概念，可以参考：Java 基本数据类型

容器框架

Java 容器框架主要分为 Collection 和 Map 两种。其中，Collection 又分为 List、Set 以及 Queue。

• Collection - 一个独立元素的序列，这些元素都服从一条或者多条规则。
  • List - 必须按照插入的顺序保存元素。常见 List 容器有：
    • ArrayList - Object[] 数组。
    • LinkedList - 双链表 (JDK1.6 之前为循环链表，JDK1.7 取消了循环）。
    • Vector - 通过 synchronized 修饰读写方法来保证并发安全。
    • Vector - Object[] 数组，通过 synchronized 修饰读写方法来保证并发安全。
  • Set - 不能有重复的元素。常见 Set 容器有：
    • HashSet - 无序，内部基于 HashMap 来实现的。
    • LinkedHashSet - 保证插入顺序，内部基于 LinkedHashMap 来实现的。
    • TreeSet - 保证自然序或用户指定的比较器顺序，内部基于红黑树实现。
  • Queue - 按照排队规则来确定对象产生的顺序。
    • PriorityQueue - 基于 Object[] 数组来实现小顶堆
    • DelayQueue - 延迟队列。
    • ArrayQueue - ArrayDeque 是 Deque 的顺序表实现。基于动态数组实现了栈和队列所需的所有操作。
    • LinkedList - LinkedList 是 Deque 的链表实现。
• Map - 一组成对的“键值对”对象，允许你使用键来查找值。常见的 Map 容器有：
  • HashMap：JDK1.8 之前 HashMap 由数组+链表组成的，数组是 HashMap 的主体，链表则是主要为了解决哈希冲突而存在的（“拉链法”解决冲突）。JDK1.8 以后在解决哈希冲突时有了较大的变化，当链表长度大于阈值（默认为 8）（将链表转换成红黑树前会判断，如果当前数组的长度小于 64，那么会选择先进行数组扩容，而不是转换为红黑树）时，将链表转化为红黑树，以减少搜索时间。
  • LinkedHashMap：LinkedHashMap 继承自 HashMap，所以它的底层仍然是基于拉链式散列结构即由数组和链表或红黑树组成。另外，LinkedHashMap 在上面结构的基础上，增加了一条双向链表，使得上面的结构可以保持键值对的插入顺序。同时通过对链表进行相应的操作，实现了访问顺序相关逻辑。
  • Hashtable：数组+链表组成的，数组是 Hashtable 的主体，链表则是主要为了解决哈希冲突而存在的。
  • TreeMap：红黑树（自平衡的排序二叉树）。

容器的基本机制

Java 的容器具有一定的共性，它们或全部或部分依赖以下技术。所以，学习以下技术点，对于理解 Java 容器的特性和原理有很大的帮助。

泛型

Java 1.5 引入了泛型技术。

Java 容器通过泛型技术来保证其数据的类型安全。什么是类型安全呢？

举例来说：如果有一个 List<Object> 容器，Java 编译器在编译时不会对原始类型进行类型安全检查，却会对带参数的类型进行检查，通过使用 Object 作为类型，可以告知编译器该方法可以接受任何类型的对象，比如 String 或 Integer。

1
2
3

List<Object> list = new ArrayList<Object>();
list.add("123");
list.add(123);

如果没有泛型技术，如示例中的代码那样，容器中就可能存储任意数据类型，这是很危险的行为。

1
2
3

List<String> list = new ArrayList<String>();
list.add("123");
list.add(123);

:bulb: 想深入了解 Java 泛型技术的用法和原理可以参考：深入理解 Java 泛型

Iterable 和 Iterator

Iterable 和 Iterator 目的在于遍历访问容器中的元素。

Iterator 接口定义：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16

public interface Iterator<E> {

    boolean hasNext();

    E next();

    default void remove() {
        throw new UnsupportedOperationException("remove");
    }

    default void forEachRemaining(Consumer<? super E> action) {
        Objects.requireNonNull(action);
        while (hasNext())
            action.accept(next());
    }
}

Iterable 接口定义：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

public interface Iterable<T> {

    Iterator<T> iterator();

    default void forEach(Consumer<? super T> action) {
        Objects.requireNonNull(action);
        for (T t : this) {
            action.accept(t);
        }
    }

    default Spliterator<T> spliterator() {
        return Spliterators.spliteratorUnknownSize(iterator(), 0);
    }
}

Collection 接口扩展了 Iterable 接口。

迭代其实我们可以简单地理解为遍历，是一个标准化遍历各类容器里面的所有对象的接口。它是一个经典的设计模式——迭代器模式（Iterator）。

迭代器模式 - 提供一种方法顺序访问一个聚合对象中各个元素，而又无须暴露该对象的内部表示。

示例：迭代器遍历

1
2
3
4
5
6
7
8
9
10
11
12
13
14

public class IteratorDemo {

    public static void main(String[] args) {
        List<Integer> list = new ArrayList<>();
        list.add(1);
        list.add(2);
        list.add(3);
        Iterator it = list.iterator();
        while (it.hasNext()) {
            System.out.println(it.next());
        }
    }

}

《阿里巴巴 Java 开发手册》的描述如下：

不要在 foreach 循环里进行元素的 remove/add 操作。remove 元素请使用 Iterator 方式，如果并发操作，需要对 Iterator 对象加锁。

通过反编译你会发现 foreach 语法底层其实还是依赖 Iterator 。不过， remove/add 操作直接调用的是集合自己的方法，而不是 Iterator 的 remove/add方法

这就导致 Iterator 莫名其妙地发现自己有元素被 remove/add ，然后，它就会抛出一个 ConcurrentModificationException 来提示用户发生了并发修改异常。这就是单线程状态下产生的 fail-fast 机制。

fail-fast 机制：多个线程对 fail-fast 集合进行修改的时候，可能会抛出ConcurrentModificationException。 即使是单线程下也有可能会出现这种情况，上面已经提到过。

相关阅读：什么是 fail-fastopen in new window 。

Java8 开始，可以使用 Collection#removeIf()方法删除满足特定条件的元素，如：

1
2
3
4
5
6

List<Integer> list = new ArrayList<>();
for (int i = 1; i <= 10; ++i) {
    list.add(i);
}
list.removeIf(filter -> filter % 2 == 0); /* 删除list中的所有偶数 */
System.out.println(list); /* [1, 3, 5, 7, 9] */

除了上面介绍的直接使用 Iterator 进行遍历操作之外，你还可以：

• 使用普通的 for 循环
• 使用 fail-safe 的集合类。java.util包下面的所有的集合类都是 fail-fast 的，而java.util.concurrent包下面的所有的类都是 fail-safe 的。
• … …

Comparable 和 Comparator

Comparable 接口和 Comparator 接口一般用于实现容器中元素的比较及排序：

• Comparable 接口实际上是出自 java.lang 包 它有一个 compareTo(Object obj) 方法用来排序
• Comparator 接口实际上是出自 java.util 包它有一个 compare(Object obj1, Object obj2) 方法用来排序

::: tabs#Comparable和Comparator接口定义

@tab Comparable 接口定义

Comparable 接口定义

1
2
3

public interface Comparable<T> {
    public int compareTo(T o);
}

@tab Comparator 接口定义

Comparator 接口定义

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

@FunctionalInterface
public interface Comparator<T> {

    int compare(T o1, T o2);

    boolean equals(Object obj);

    // 反转
    default Comparator<T> reversed() {
        return Collections.reverseOrder(this);
    }

    default Comparator<T> thenComparing(Comparator<? super T> other) {
        Objects.requireNonNull(other);
        return (Comparator<T> & Serializable) (c1, c2) -> {
            int res = compare(c1, c2);
            return (res != 0) ? res : other.compare(c1, c2);
        };
    }

    // thenComparingXXX 方法略

    // 静态方法略
}

:::

假设，有一个 List 容器，存储的是 User 类型对象。现在要根据 User 中的 age 属性进行排序。

User 定义如下：

1
2
3
4
5
6
7
8
9
10
11

public class User {

    private String name;
    private int age;

    public User(String name, int age) {
        this.age = age;
        this.name = name;
    }
    // getter、setter 略
}

我们分别通过 Comparable 和 Comparator 来实现比较、排序，体会一下有何差异。

::: tabs#Comparable和Comparator使用示例

@tab Comparable 接口使用示例

Comparable 接口使用示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

public class ComparableDemo {

    public static void main(String[] args) {
        User a = new User("A", 18);
        User b = new User("B", 17);
        User c = new User("C", 20);
        List<User> list = new ArrayList<>(Arrays.asList(a, b, c));
        Collections.sort(list);
        list.forEach(System.out::println);
    }
    // 输出：
    // User{age=17, name='B'}
    // User{age=18, name='A'}
    // User{age=20, name='C'}

    // 需要对被比较、排序的类进行改造，实现 Comparable 接口
    static class User implements Comparable<User> {

        private String name;
        private int age;

        public User(String name, int age) {
            this.age = age;
            this.name = name;
        }
        
        // getter、setter 略

        @Override
        public int compareTo(User o) {
            return this.age - o.age;
        }

        @Override
        public String toString() {
            return "User{" + "age=" + age + ", name='" + name + '\'' + '}';
        }
    }
}

从上例可以看出，使用 Comparable 接口，被排序对象类必须实现 Comparable 接口；并在类中定义 compareTo 方法的实现，即排序逻辑必须置于被排序对象类中。

@tab Comparator 接口使用示例

Comparator 接口使用示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

public class ComparatorDemo {

    public static void main(String[] args) {
        User a = new User("A", 18);
        User b = new User("B", 17);
        User c = new User("C", 20);
        List<User> list = new ArrayList<>(Arrays.asList(a, b, c));
        Collections.sort(list, new Comparator<User>() {
            @Override
            public int compare(User o1, User o2) {
                return o1.age - o2.age;
            }
        });
        list.forEach(System.out::println);
    }
    // 输出：
    // User{age=17, name='B'}
    // User{age=18, name='A'}
    // User{age=20, name='C'}

    static class User {

        private String name;
        private int age;

        public User(String name, int age) {
            this.age = age;
            this.name = name;
        }

        // getter、setter 略

        @Override
        public String toString() {
            return "User{" + "age=" + age + ", name='" + name + '\'' + '}';
        }
    }
}

从上例可以看出，使用 Comparator 接口和 Comparable 接口的不同点在于：被排序的对象类无需实现 Comparator 接口，排序逻辑置于被排序对象类的外部。

:::

Cloneable

Java 中 一个类要实现 clone 功能 必须实现 Cloneable 接口，否则在调用 clone() 时会报 CloneNotSupportedException 异常。

Java 中所有类都默认继承 java.lang.Object 类，在 java.lang.Object 类中有一个方法 clone()，这个方法将返回 Object 对象的一个拷贝。Object 类里的 clone() 方法仅仅用于浅拷贝（拷贝基本成员属性，对于引用类型仅返回指向改地址的引用）。

如果 Java 类需要深拷贝，需要覆写 clone() 方法。

fail-fast

fail-fast 的要点

Java 容器（如：ArrayList、HashMap、TreeSet 等待）的 javadoc 中常常提到类似的描述：

注意，迭代器的快速失败行为无法得到保证，因为一般来说，不可能对是否出现不同步并发修改做出任何硬性保证。快速失败（fail-fast）迭代器会尽最大努力抛出 ConcurrentModificationException。因此，为提高这类迭代器的正确性而编写一个依赖于此异常的程序是错误的做法：迭代器的快速失败行为应该仅用于检测 bug。

那么，我们不禁要问，什么是 fail-fast，为什么要有 fail-fast 机制？

fail-fast 是 Java 容器的一种错误检测机制。当多个线程对容器进行结构上的改变的操作时，就可能触发 fail-fast 机制。记住是有可能，而不是一定。

例如：假设存在两个线程（线程 1、线程 2），线程 1 通过 Iterator 在遍历容器 A 中的元素，在某个时候线程 2 修改了容器 A 的结构（是结构上面的修改，而不是简单的修改容器元素的内容），那么这个时候程序就会抛出 ConcurrentModificationException 异常，从而产生 fail-fast 机制。

容器在迭代操作中改变元素个数（添加、删除元素）都可能会导致 fail-fast。

示例：fail-fast 示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51

public class FailFastDemo {

    private static int MAX = 100;

    private static List<Integer> list = new ArrayList<>();

    public static void main(String[] args) {
        for (int i = 0; i < MAX; i++) {
            list.add(i);
        }
        new Thread(new MyThreadA()).start();
        new Thread(new MyThreadB()).start();
    }

    /** 迭代遍历容器所有元素 */
    static class MyThreadA implements Runnable {

        @Override
        public void run() {
            Iterator<Integer> iterator = list.iterator();
            while (iterator.hasNext()) {
                int i = iterator.next();
                System.out.println("MyThreadA 访问元素：" + i);
                try {
                    TimeUnit.MILLISECONDS.sleep(100);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }

    }

    /** 遍历删除指定范围内的所有偶数 */
    static class MyThreadB implements Runnable {

        @Override
        public void run() {
            int i = 0;
            while (i < MAX) {
                if (i % 2 == 0) {
                    System.out.println("MyThreadB 删除元素" + i);
                    list.remove(i);
                }
                i++;
            }
        }

    }

}

执行后，会抛出 java.util.ConcurrentModificationException 异常。

解决 fail-fast

fail-fast 有两种解决方案：

• 在遍历过程中所有涉及到改变容器个数的地方全部加上 synchronized 或者直接使用 Collections.synchronizedXXX 容器，这样就可以解决。但是不推荐，因为增删造成的同步锁可能会阻塞遍历操作，影响吞吐。
• 使用并发容器，如：CopyOnWriterArrayList。

容器和线程安全

为了在并发环境下安全地使用容器，Java 提供了同步容器和并发容器。

同步容器和并发容器详情请参考：Java 并发之容器

参考资料

• Java 编程思想（第 4 版）
• 由浅入深理解 java 集合（一）——集合框架 Collection、Map
• 由浅入深理解 java 集合（二）——集合 Set
• Java 提高篇（三十）—–Iterator
• Java 提高篇（三四）—–fail-fast 机制

Java 并发之 AQS

AQS 简介

AQS 是 AbstractQueuedSynchronizer 的缩写，即 队列同步器，顾名思义，其主要作用是处理同步。它是并发锁和很多同步工具类的实现基石（如 ReentrantLock、ReentrantReadWriteLock、CountDownLatch、Semaphore、FutureTask 等）。

**AQS 提供了对锁和同步器的通用能力支持 **。在 java.util.concurrent.locks 包中的相关锁（常用的有 ReentrantLock、 ThreadPoolExecutor）都是基于 AQS 来实现。这些锁都没有直接继承 AQS，而是定义了一个 Sync 类去继承 AQS。为什么要这样呢？因为锁面向的是使用用户，而同步器面向的则是线程控制，那么在锁的实现中聚合同步器而不是直接继承 AQS 就可以很好的隔离二者所关注的事情。

AQS 的应用

AQS 定义两种资源共享方式：Exclusive（独占，只有一个线程能执行，如 ReentrantLock）和 Share（共享，多个线程可同时执行，如 Semaphore / CountDownLatch）。

独占锁 API

获取、释放独占锁的主要 API 如下：

1
2
3
4

public final void acquire(int arg)
public final void acquireInterruptibly(int arg)
public final boolean tryAcquireNanos(int arg, long nanosTimeout)
public final boolean release(int arg)

• acquire - 获取独占锁。
• acquireInterruptibly - 获取可中断的独占锁。
• tryAcquireNanos - 尝试在指定时间内获取可中断的独占锁。在以下三种情况下回返回：
  • 在超时时间内，当前线程成功获取了锁；
  • 当前线程在超时时间内被中断；
  • 超时时间结束，仍未获得锁返回 false。
• release - 释放独占锁。

共享锁 API

获取、释放共享锁的主要 API 如下：

1
2
3
4

public final void acquireShared(int arg)
public final void acquireSharedInterruptibly(int arg)
public final boolean tryAcquireSharedNanos(int arg, long nanosTimeout)
public final boolean releaseShared(int arg)

• acquireShared - 获取共享锁。
• acquireSharedInterruptibly - 获取可中断的共享锁。
• tryAcquireSharedNanos - 尝试在指定时间内获取可中断的共享锁。
• release - 释放共享锁。

AQS 的原理

AQS 核心思想是，如果被请求的共享资源空闲，则将当前请求资源的线程设置为有效的工作线程，并且将共享资源设置为锁定状态；如果被请求的共享资源被占用，那么就需要一套线程阻塞等待以及被唤醒时锁分配的机制。这个机制是基于 CLH 锁 （Craig, Landin, and Hagersten locks） 的变体实现的，将暂时获取不到锁的线程加入到队列中。

CLH 本是一个单向队列，AQS 中的队列采用了 CLH 的变体，是一个虚拟的 FIFO 双向队列（虚拟的双向队列，是指不存在结点实例，仅存在结点之间的关联关系），暂时获取不到锁的线程将被加入到该队列中。AQS 将每条请求共享资源的线程封装成一个 CLH 队列锁的一个结点（Node）来实现锁的分配。在 CLH 队列锁中，一个节点表示一个线程，它保存着线程的引用（thread）、 当前节点在队列中的状态（waitStatus）、前驱节点（prev）、后继节点（next）。

AQS 的核心原理图：

AQS 的数据结构

先看一下 AbstractQueuedSynchronizer 的定义：

1
2
3
4
5
6
7
8
9
10
11

public abstract class AbstractQueuedSynchronizer
    extends AbstractOwnableSynchronizer
    implements java.io.Serializable {

    /** 等待队列的队头，懒加载。只能通过 setHead 方法修改。 */
    private transient volatile Node head;
    /** 等待队列的队尾，懒加载。只能通过 enq 方法添加新的等待节点。*/
    private transient volatile Node tail;
    /** 同步状态 */
    private volatile int state;
}

阅读 AQS 的源码，可以发现：AQS 继承自 AbstractOwnableSynchronize，它有以下核心属性：

• state - AQS 使用一个整型的 volatile 变量来 维护同步状态。这个整数状态的意义由子类来赋予，如 ReentrantLock 中该状态值表示所有者线程已经重复获取该锁的次数；Semaphore 中该状态值表示剩余的许可数量。
• head 和 tail - AQS **维护了一个 Node 类型（AQS 的内部类）的双向队列来完成同步状态的管理 **。这个双向队列是一个双向的 FIFO 队列，通过 head 和 tail 指针进行访问。当 **有线程获取锁失败后，就被添加到队列末尾 **。

再来看一下 Node 的源码，很显然，Node 是一个双向队列结构：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23

static final class Node {
    /** 该等待同步的节点处于共享模式 */
    static final Node SHARED = new Node();
    /** 该等待同步的节点处于独占模式 */
    static final Node EXCLUSIVE = null;

    /** 线程等待状态，状态值有：0、1、-1、-2、-3 */
    volatile int waitStatus;
    static final int CANCELLED =  1;
    static final int SIGNAL    = -1;
    static final int CONDITION = -2;
    static final int PROPAGATE = -3;

    /** 前驱节点 */
    volatile Node prev;
    /** 后继节点 */
    volatile Node next;
    /** 等待锁的线程 */
    volatile Thread thread;

  	/** 和节点是否共享有关 */
    Node nextWaiter;
}

属性说明：

waitStatus 是一个整型的 volatile 变量，用来维护 AQS 同步队列中线程节点的状态。waitStatus 有五个状态值：

• 0 - 一个 Node 被初始化的时候的默认值
• CANCELLED(1) - 表示线程获取锁的请求已经取消了
• SIGNAL(-1) - 表示线程已经准备好了，就等资源释放了
• CONDITION(-2) - 表示节点在等待队列中，节点线程等待唤醒
• PROPAGATE(-3) - 当前线程处在 SHARED 情况下，该字段才会使用

独占锁的获取和释放

获取独占锁

AQS 中使用 acquire(int arg) 方法获取独占锁的相关源码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

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

// 利用 CAS 操作将当前线程加入等待队列队尾
private Node addWaiter(Node mode) {
	Node node = new Node(Thread.currentThread(), mode);
	Node pred = tail;
	if (pred != null) {
		node.prev = pred;
		if (compareAndSetTail(pred, node)) {
			pred.next = node;
			return node;
		}
	}
	enq(node);
	return node;
}

// 自旋尝试为等待队列中的线程节点获取独占锁
final boolean acquireQueued(final Node node, int arg) {
	boolean failed = true;
	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())
				interrupted = true;
		}
	} finally {
		if (failed)
			cancelAcquire(node);
	}
}

其大致流程如下：

1. 先通过 tryAcquire 尝试获取同步状态，如果获取同步状态成功，则结束方法，直接返回。
2. 若不成功，调用 addWaiter 方法，利用 CAS 操作将当前线程加入等待队列队尾。
3. 接着，自旋尝试为等待队列中的线程节点获取独占锁，直到获取成功或线程中断。

释放独占锁

AQS 中使用 acquire(int arg) 方法获取独占锁的相关源码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28

public final boolean release(int arg) {
    // 尝试释放锁
	if (tryRelease(arg)) {
		Node h = head;
        // 如果队列不为空，唤醒下一个节点中的线程
		if (h != null && h.waitStatus != 0)
			unparkSuccessor(h);
		return true;
	}
	return false;
}

private void unparkSuccessor(Node node) {

	int ws = node.waitStatus;
	if (ws < 0)
		compareAndSetWaitStatus(node, ws, 0);

	Node s = node.next;
	if (s == null || s.waitStatus > 0) {
		s = null;
		for (Node t = tail; t != null && t != node; t = t.prev)
			if (t.waitStatus <= 0)
				s = t;
	}
	if (s != null)
		LockSupport.unpark(s.thread);
}

1. 先尝试获取解锁线程的同步状态，如果获取同步状态不成功，则结束方法，直接返回。
2. 如果获取同步状态成功且队列不为空，AQS 会尝试唤醒下一个节点中的线程。

获取可中断的独占锁

AQS 中使用 acquireInterruptibly(int arg) 方法获取可中断的独占锁。

acquireInterruptibly(int arg) 实现方式 相较于获取独占锁方法（ acquire）非常相似，区别仅在于它会 通过 Thread.interrupted 检测当前线程是否被中断，如果是，则立即抛出中断异常（InterruptedException）。

限时获取独占锁

AQS 中使用 tryAcquireNanos(int arg) 方法获取超时等待的独占锁。

doAcquireNanos 的实现方式 相较于获取独占锁方法（ acquire）非常相似，区别在于它会根据超时时间和当前时间计算出截止时间。在获取锁的流程中，会不断判断是否超时，如果超时，直接返回 false；如果没超时，则用 LockSupport.parkNanos 来阻塞当前线程。

共享锁的获取和释放

获取共享锁

AQS 中使用 acquireShared(int arg) 方法获取共享锁。

acquireShared 方法和 acquire 方法的逻辑很相似，区别仅在于自旋的条件以及节点出队的操作有所不同。

成功获得共享锁的条件如下：

• tryAcquireShared(arg) 返回值大于等于 0 （这意味着共享锁的 permit 还没有用完）。
• 当前节点的前驱节点是头结点。

释放共享锁

AQS 中使用 releaseShared(int arg) 方法释放共享锁。

releaseShared 首先会尝试释放同步状态，如果成功，则解锁一个或多个后继线程节点。释放共享锁和释放独占锁流程大体相似，区别在于：

对于独占模式，如果需要 SIGNAL，释放仅相当于调用头节点的 unparkSuccessor。

获取可中断的共享锁

AQS 中使用 acquireSharedInterruptibly(int arg) 方法获取可中断的共享锁。

acquireSharedInterruptibly 方法与 acquireInterruptibly 几乎一致，不再赘述。

限时获取共享锁

AQS 中使用 tryAcquireSharedNanos(int arg) 方法获取超时等待式的共享锁。

tryAcquireSharedNanos 方法与 tryAcquireNanos 几乎一致，不再赘述。

自定义同步器

同步器的设计是基于模板方法模式的，如果需要自定义同步器一般的方式是这样（模板方法模式很经典的一个应用）：

1. 使用者继承 AbstractQueuedSynchronizer 并重写指定的方法。
2. 将 AQS 组合在自定义同步组件的实现中，并调用其模板方法，而这些模板方法会调用使用者重写的方法。

这和我们以往通过实现接口的方式有很大区别，这是模板方法模式很经典的一个运用。

AQS 使用了模板方法模式，自定义同步器时需要重写下面几个 AQS 提供的钩子方法：

1
2
3
4
5
6
7
8
9
10

// 独占方式。尝试获取资源，成功则返回 true，失败则返回 false。
protected boolean tryAcquire(int)
// 独占方式。尝试释放资源，成功则返回 true，失败则返回 false。
protected boolean tryRelease(int)
// 共享方式。尝试获取资源。负数表示失败；0 表示成功，但没有剩余可用资源；正数表示成功，且有剩余资源。
protected int tryAcquireShared(int)
// 共享方式。尝试释放资源，成功则返回 true，失败则返回 false。
protected boolean tryReleaseShared(int)
// 该线程是否正在独占资源。只有用到 condition 才需要去实现它。
protected boolean isHeldExclusively()

什么是钩子方法呢？ 钩子方法是一种被声明在抽象类中的方法，一般使用 protected 关键字修饰，它可以是空方法（由子类实现），也可以是默认实现的方法。模板设计模式通过钩子方法控制固定步骤的实现。

参考资料

• 《Java 并发编程实战》
• 《Java 并发编程的艺术》
• 极客时间教程 - Java 并发编程实战
• Java 并发编程：Lock
• 深入学习 java 同步器 AQS
• AbstractQueuedSynchronizer 框架
• Java 中的锁分类

Java 并发之锁

本文先阐述 Java 中各种锁的概念。

然后，重点介绍 Lock 和 Condition 两个接口及其实现。并发编程有两个核心问题：同步和互斥。

互斥，即同一时刻只允许一个线程访问共享资源；

同步，即线程之间如何通信、协作。

这两大问题，管程（sychronized）都是能够解决的。J.U.C 包还提供了 Lock 和 Condition 两个接口来实现管程，其中 Lock 用于解决互斥问题，Condition 用于解决同步问题。

并发锁简介

确保线程安全最常见的做法是利用锁机制（Lock、sychronized）来对共享数据做互斥同步，这样在同一个时刻，只有一个线程可以执行某个方法或者某个代码块，那么操作必然是原子性的，线程安全的。

在工作、面试中，经常会听到各种五花八门的锁，听的人云里雾里。锁的概念术语很多，它们是针对不同的问题所提出的，通过简单的梳理，也不难理解。

可重入锁

可重入锁，顾名思义，指的是线程可以重复获取同一把锁。即同一个线程在外层方法获取了锁，在进入内层方法会自动获取锁。

可重入锁可以在一定程度上避免死锁。

• ReentrantLock 、ReentrantReadWriteLock 是可重入锁。这点，从其命名也不难看出。
• synchronized 也是一个可重入锁。

【示例】synchronized 的可重入示例

1
2
3
4
5
6
7
8

synchronized void setA() throws Exception{
    Thread.sleep(1000);
    setB();
}

synchronized void setB() throws Exception{
    Thread.sleep(1000);
}

上面的代码就是一个典型场景：如果使用的锁不是可重入锁的话，setB 可能不会被当前线程执行，从而造成死锁。

【示例】ReentrantLock 的可重入示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34

class Task {

    private int value;
    private final Lock lock = new ReentrantLock();

    public Task() {
        this.value = 0;
    }

    public int get() {
        // 获取锁
        lock.lock();
        try {
            return value;
        } finally {
            // 保证锁能释放
            lock.unlock();
        }
    }

    public void addOne() {
        // 获取锁
        lock.lock();
        try {
            // 注意：此处已经成功获取锁，进入 get 方法后，又尝试获取锁，
            // 如果锁不是可重入的，会导致死锁
            value = 1 + get();
        } finally {
            // 保证锁能释放
            lock.unlock();
        }
    }

}

公平锁与非公平锁

• 公平锁 - 公平锁是指 多线程按照申请锁的顺序来获取锁。
• 非公平锁 - 非公平锁是指 多线程不按照申请锁的顺序来获取锁 。这就可能会出现优先级反转（后来者居上）或者饥饿现象（某线程总是抢不过别的线程，导致始终无法执行）。

公平锁为了保证线程申请顺序，势必要付出一定的性能代价，因此其吞吐量一般低于非公平锁。

公平锁与非公平锁 在 Java 中的典型实现：

• synchronized 只支持非公平锁。
• ReentrantLock 、ReentrantReadWriteLock，默认是非公平锁，但支持公平锁。

独占锁与共享锁

独占锁与共享锁是一种广义上的说法，从实际用途上来看，也常被称为互斥锁与读写锁。

• 独占锁 - 独占锁是指 锁一次只能被一个线程所持有。
• 共享锁 - 共享锁是指 锁可被多个线程所持有。

独占锁与共享锁在 Java 中的典型实现：

• synchronized 、ReentrantLock 只支持独占锁。
• ReentrantReadWriteLock 其写锁是独占锁，其读锁是共享锁。读锁是共享锁使得并发读是非常高效的，读写，写读 ，写写的过程是互斥的。

悲观锁与乐观锁

乐观锁与悲观锁不是指具体的什么类型的锁，而是处理并发同步的策略。

悲观锁（Pessimistic Lock）

• 总是假设最坏的情况，认为：不加锁的并发操作一定会出问题。
• 悲观锁在 Java 中的应用就是通过使用 synchronized 和 Lock 显示加锁来进行互斥同步，这是一种阻塞同步。
• 悲观锁适合写操作频繁的场景。高并发的场景下，激烈的锁竞争会造成线程阻塞，大量阻塞线程会导致系统的上下文切换，增加系统的性能开销。并且，悲观锁还可能会存在死锁问题，影响代码的正常运行。

【示例】悲观锁示例

1
2
3
4
5
6
7
8
9
10
11
12
13

public void syncTask() {
    synchronized (this) {
        // 需要同步的操作
    }
}

private Lock lock = new ReentrantLock();
lock.lock();
try {
   // 需要同步的操作
} finally {
    lock.unlock();
}

乐观锁（OptimisticLock）

• 乐观锁总是假设最好的情况，认为：不加锁的并发操作也没什么问题。每次访问数据时，都假设数据不会被其他线程修改，不必加锁。虽然不加锁，但不意味着什么都不做，而是在更新的时候，判断一下在此期间是否有其他线程更新该数据。
• 乐观锁最常见的实现方式，是使用版本号机制或 CAS 算法（Compare And Swap）去实现。Java 中的原子类就是基于 CAS 实现的。
• 乐观锁的优点是：减少锁竞争，提高并发度。
• 乐观锁的缺点是：
  • 存在 ABA 问题。所谓的 ABA 问题是指在并发编程中，如果一个变量初次读取的时候是 A 值，它的值被改成了 B，然后又其他线程把 B 值改成了 A，而另一个早期线程在对比值时会误以为此值没有发生改变，但其实已经发生变化了
  • 如果乐观锁所检查的数据存在大量锁竞争，会由于不断循环重试，产生大量的 CPU 开销。
• 乐观锁适合读多写少的场景。高并发的场景下，乐观锁相比悲观锁来说，不存在锁竞争造成线程阻塞，也不会有死锁的问题，在性能上往往会更胜一筹。但是，如果冲突频繁发生（写占比非常多的情况），会频繁失败和重试，这样同样会非常影响性能，导致 CPU 飙升。

【示例】乐观锁示例

1
2
3
4
5
6
7
8
9
10

// AtomicInteger 的 getAndAccumulate 方法采用了自旋 + CAS 的乐观锁模式
public final int getAndAccumulate(int x,
	IntBinaryOperator accumulatorFunction) {
	int prev, next;
	do {
		prev = get();
		next = accumulatorFunction.applyAsInt(prev, x);
	} while (!compareAndSet(prev, next));
	return prev;
}

乐观锁也是一种通用的锁机制，不仅在 Java 中，在其他很多软件领域，也存在乐观锁机制。比如下面的示例是 MySQL 中的乐观锁示例。

假设，order 表中有一个字段 status，表示订单状态：status 为 1 代表订单未支付；status 为 2 代表订单已支付。现在，要将 id 为 1 的订单状态置为已支付，则操作如下：

1
2
3
4
5

select status, version from order where id=#{id}

update order
set status=2, version=version+1
where id=#{id} and version=#{version};

偏向锁、轻量级锁、重量级锁

所谓轻量级锁与重量级锁，指的是锁控制粒度的粗细。显然，控制粒度越细，阻塞开销越小，并发性也就越高。

Java 1.6 以前，重量级锁一般指的是 synchronized ，而轻量级锁指的是 volatile。

Java 1.6 以后，针对 synchronized 做了大量优化，引入 4 种锁状态： 无锁状态、偏向锁、轻量级锁和重量级锁。锁可以单向的从偏向锁升级到轻量级锁，再从轻量级锁升级到重量级锁 。

• 偏向锁 - 偏向锁是指一段同步代码一直被一个线程所访问，那么该线程会自动获取锁。降低获取锁的代价。
• 轻量级锁 - 是指当锁是偏向锁的时候，被另一个线程所访问，偏向锁就会升级为轻量级锁，其他线程会通过自旋的形式尝试获取锁，不会阻塞，提高性能。
• 重量级锁 - 是指当锁为轻量级锁的时候，另一个线程虽然是自旋，但自旋不会一直持续下去，当自旋一定次数的时候，还没有获取到锁，就会进入阻塞，该锁膨胀为重量级锁。重量级锁会让其他申请的线程进入阻塞，性能降低。

分段锁

分段锁其实是一种锁的设计，并不是具体的一种锁。所谓分段锁，就是把锁的对象分成多段，每段独立控制，使得锁粒度更细，减少阻塞开销，从而提高并发性。这其实很好理解，就像高速公路上的收费站，如果只有一个收费口，那所有的车只能排成一条队缴费；如果有多个收费口，就可以分流了。

Hashtable 使用 synchronized 修饰方法来保证线程安全性，那么面对线程的访问，Hashtable 就会锁住整个对象，所有的其它线程只能等待，这种阻塞方式的吞吐量显然很低。

Java 1.7 以前的 ConcurrentHashMap 就是分段锁的典型案例。ConcurrentHashMap 维护了一个 Segment 数组，一般称为分段桶。

1

final Segment<K,V>[] segments;

当有线程访问 ConcurrentHashMap 的数据时，ConcurrentHashMap 会先根据 hashCode 计算出数据在哪个桶（即哪个 Segment），然后锁住这个 Segment。

内置锁和显示锁

Java 1.5 之前，协调对共享对象的访问时可以使用的机制只有 synchronized 和 volatile。这两个都属于内置锁，即锁的申请和释放都是由 JVM 所控制。

Java 1.5 之后，增加了新的机制：ReentrantLock、ReentrantReadWriteLock ，这类锁的申请和释放都可以由程序所控制，所以常被称为显示锁。

💡 synchronized 的用法和原理可以参考：Java 并发基础机制 - synchronized 。

:bell: 注意：如果不需要 ReentrantLock、ReentrantReadWriteLock 所提供的高级同步特性，**应该优先考虑使用 synchronized**。理由如下：

• Java 1.6 以后，synchronized 做了大量的优化，其性能已经与 ReentrantLock、ReentrantReadWriteLock 基本上持平。
• 从趋势来看，Java 未来更可能会优化 synchronized ，而不是 ReentrantLock、ReentrantReadWriteLock ，因为 synchronized 是 JVM 内置属性，它能执行一些优化。
• ReentrantLock、ReentrantReadWriteLock 申请和释放锁都是由程序控制，如果使用不当，可能造成死锁，这是很危险的。

以下对比一下显示锁和内置锁的差异：

• 主动获取锁和释放锁
  • synchronized 不能主动获取锁和释放锁。获取锁和释放锁都是 JVM 控制的。
  • ReentrantLock 可以主动获取锁和释放锁。（如果忘记释放锁，就可能产生死锁）。
• 响应中断
  • synchronized 不能响应中断。
  • ReentrantLock 可以响应中断。
• 超时机制
  • synchronized 没有超时机制。
  • ReentrantLock 有超时机制。ReentrantLock 可以设置超时时间，超时后自动释放锁，避免一直等待。
• 支持公平锁
  • synchronized 只支持非公平锁。
  • ReentrantLock 支持非公平锁和公平锁。
• 是否支持共享
  • 被 synchronized 修饰的方法或代码块，只能被一个线程访问（独享）。如果这个线程被阻塞，其他线程也只能等待
  • ReentrantLock 可以基于 Condition 灵活的控制同步条件。
• 是否支持读写分离
  • synchronized 不支持读写锁分离；
  • ReentrantReadWriteLock 支持读写锁，从而使阻塞读写的操作分开，有效提高并发性。

Lock 和 Condition

为何引入 Lock 和 Condition

并发编程领域，有两大核心问题：一个是互斥，即同一时刻只允许一个线程访问共享资源；另一个是同步，即线程之间如何通信、协作。这两大问题，管程都是能够解决的。Java SDK 并发包通过 Lock 和 Condition 两个接口来实现管程，其中 Lock 用于解决互斥问题，Condition 用于解决同步问题。

synchronized 是管程的一种实现，既然如此，何必再提供 Lock 和 Condition。

JDK 1.6 以前，synchronized 还没有做优化，性能远低于 Lock。但是，性能不是引入 Lock 的最重要因素。真正关键在于：synchronized 使用不当，可能会出现死锁。synchronized 无法通过破坏不可抢占条件来避免死锁。原因是 synchronized 申请资源的时候，如果申请不到，线程直接进入阻塞状态了，而线程进入阻塞状态，啥都干不了，也释放不了线程已经占有的资源。

与内置锁 synchronized 不同的是，**Lock 提供了一组无条件的、可轮询的、定时的以及可中断的锁操作**，所有获取锁、释放锁的操作都是显式的操作。

• 能够响应中断。synchronized 的问题是，持有锁 A 后，如果尝试获取锁 B 失败，那么线程就进入阻塞状态，一旦发生死锁，就没有任何机会来唤醒阻塞的线程。但如果阻塞状态的线程能够响应中断信号，也就是说当我们给阻塞的线程发送中断信号的时候，能够唤醒它，那它就有机会释放曾经持有的锁 A。这样就破坏了不可抢占条件了。
• 支持超时。如果线程在一段时间之内没有获取到锁，不是进入阻塞状态，而是返回一个错误，那这个线程也有机会释放曾经持有的锁。这样也能破坏不可抢占条件。
• 非阻塞地获取锁。如果尝试获取锁失败，并不进入阻塞状态，而是直接返回，那这个线程也有机会释放曾经持有的锁。这样也能破坏不可抢占条件。

Lock 接口

Lock 的接口定义如下：

1
2
3
4
5
6
7
8

public interface Lock {
    void lock();
    void lockInterruptibly() throws InterruptedException;
    boolean tryLock();
    boolean tryLock(long time, TimeUnit unit) throws InterruptedException;
    void unlock();
    Condition newCondition();
}

• lock() - 获取锁。
• unlock() - 释放锁。
• tryLock() - 尝试获取锁，仅在调用时锁未被另一个线程持有的情况下，才获取该锁。
• tryLock(long time, TimeUnit unit) - 和 tryLock() 类似，区别仅在于限定时间，如果限定时间内未获取到锁，视为失败。
• lockInterruptibly() - 锁未被另一个线程持有，且线程没有被中断的情况下，才能获取锁。
• newCondition() - 返回一个绑定到 Lock 对象上的 Condition 实例。

Condition

Condition 实现了管程模型里面的条件变量。

前文中提过 Lock 接口中 有一个 newCondition() 方法用于返回一个绑定到 Lock 对象上的 Condition 实例。Condition 是什么？有什么作用？本节将一一讲解。

在单线程中，一段代码的执行可能依赖于某个状态，如果不满足状态条件，代码就不会被执行（典型的场景，如：if ... else ...）。在并发环境中，当一个线程判断某个状态条件时，其状态可能是由于其他线程的操作而改变，这时就需要有一定的协调机制来确保在同一时刻，数据只能被一个线程锁修改，且修改的数据状态被所有线程所感知。

Java 1.5 之前，主要是利用 Object 类中的 wait、notify、notifyAll 配合 synchronized 来进行线程间通信。wait、notify、notifyAll 需要配合 synchronized 使用，不适用于 Lock。而使用 Lock 的线程，彼此间通信应该使用 Condition 。这可以理解为，什么样的锁配什么样的钥匙。内置锁（synchronized）配合内置条件队列（wait、notify、notifyAll ），显式锁（Lock）配合显式条件队列（Condition ）。

Condition 的特性

Condition 接口定义如下：

1
2
3
4
5
6
7
8
9

public interface Condition {
    void await() throws InterruptedException;
    void awaitUninterruptibly();
    long awaitNanos(long nanosTimeout) throws InterruptedException;
    boolean await(long time, TimeUnit unit) throws InterruptedException;
    boolean awaitUntil(Date deadline) throws InterruptedException;
    void signal();
    void signalAll();
}

其中，await、signal、signalAll 与 wait、notify、notifyAll 相对应，功能也相似。除此以外，Condition 相比内置条件队列（ wait、notify、notifyAll ），提供了更为丰富的功能：

• 每个锁（Lock）上可以存在多个 Condition，这意味着锁的状态条件可以有多个。
• 支持公平的或非公平的队列操作。
• 支持可中断的条件等待，相关方法：awaitUninterruptibly() 。
• 支持可定时的等待，相关方法：awaitNanos(long) 、await(long, TimeUnit)、awaitUntil(Date)。

Condition 的用法

这里以 Condition 来实现一个消费者、生产者模式。

:bell: 注意：事实上，解决此类问题使用 CountDownLatch、Semaphore 等工具更为便捷、安全。想了解详情，可以参考 Java 并发工具类 。

产品类

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59

class Message {

    private final Lock lock = new ReentrantLock();

    private final Condition producedMsg = lock.newCondition();

    private final Condition consumedMsg = lock.newCondition();

    private String message;

    private boolean state;

    private boolean end;

    public void consume() {
        //lock
        lock.lock();
        try {
            // no new message wait for new message
            while (!state) { producedMsg.await(); }

            System.out.println("consume message : " + message);
            state = false;
            // message consumed, notify waiting thread
            consumedMsg.signal();
        } catch (InterruptedException ie) {
            System.out.println("Thread interrupted - viewMessage");
        } finally {
            lock.unlock();
        }
    }

    public void produce(String message) {
        lock.lock();
        try {
            // last message not consumed, wait for it be consumed
            while (state) { consumedMsg.await(); }

            System.out.println("produce msg: " + message);
            this.message = message;
            state = true;
            // new message added, notify waiting thread
            producedMsg.signal();
        } catch (InterruptedException ie) {
            System.out.println("Thread interrupted - publishMessage");
        } finally {
            lock.unlock();
        }
    }

    public boolean isEnd() {
        return end;
    }

    public void setEnd(boolean end) {
        this.end = end;
    }

}

消费者

1
2
3
4
5
6
7
8
9
10
11
12
13
14

class MessageConsumer implements Runnable {

    private Message message;

    public MessageConsumer(Message msg) {
        message = msg;
    }

    @Override
    public void run() {
        while (!message.isEnd()) { message.consume(); }
    }

}

生产者

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33

class MessageProducer implements Runnable {

    private Message message;

    public MessageProducer(Message msg) {
        message = msg;
    }

    @Override
    public void run() {
        produce();
    }

    public void produce() {
        List<String> msgs = new ArrayList<>();
        msgs.add("Begin");
        msgs.add("Msg1");
        msgs.add("Msg2");

        for (String msg : msgs) {
            message.produce(msg);
            try {
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }

        message.produce("End");
        message.setEnd(true);
    }

}

测试

1
2
3
4
5
6
7
8
9
10

public class LockConditionDemo {

    public static void main(String[] args) {
        Message msg = new Message();
        Thread producer = new Thread(new MessageProducer(msg));
        Thread consumer = new Thread(new MessageConsumer(msg));
        producer.start();
        consumer.start();
    }
}

ReentrantLock

ReentrantLock 类是 Lock 接口的具体实现，与内置锁 synchronized 相同的是，它是一个可重入锁。

ReentrantLock 的特性如下：

• ReentrantLock 提供了与 synchronized 相同的互斥性、内存可见性和可重入性。
• ReentrantLock 支持公平锁和非公平锁（默认）两种模式。
• ReentrantLock 实现了 Lock 接口，支持了 synchronized 所不具备的灵活性，增加了轮询、超时、中断等功能。
  • synchronized 无法中断一个正在等待获取锁的线程
  • synchronized 无法在请求获取一个锁时无休止地等待

ReentrantLock 的用法

前文了解了 ReentrantLock 的特性，接下来，我们要讲述其具体用法。

ReentrantLock 的构造方法

ReentrantLock 有两个构造方法：

1
2

public ReentrantLock() {}
public ReentrantLock(boolean fair) {}

• ReentrantLock() - 默认构造方法会初始化一个非公平锁（NonfairSync）；
• ReentrantLock(boolean) - new ReentrantLock(true) 会初始化一个公平锁（FairSync）。

lock 和 unlock 方法

• lock() - 无条件获取锁。如果当前线程无法获取锁，则当前线程进入休眠状态不可用，直至当前线程获取到锁。如果该锁没有被另一个线程持有，则获取该锁并立即返回，将锁的持有计数设置为 1。
• unlock() - 用于释放锁。

:bell: 注意：请务必牢记，获取锁操作 lock() 必须在 try catch 块中进行，并且将释放锁操作 unlock() 放在 finally 块中进行，以保证锁一定被被释放，防止死锁的发生。

示例：ReentrantLock 的基本操作

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67

public class ReentrantLockDemo {

    public static void main(String[] args) {
        Task task = new Task();
        MyThread tA = new MyThread("Thread-A", task);
        MyThread tB = new MyThread("Thread-B", task);
        MyThread tC = new MyThread("Thread-C", task);
        tA.start();
        tB.start();
        tC.start();
    }

    static class MyThread extends Thread {

        private Task task;

        public MyThread(String name, Task task) {
            super(name);
            this.task = task;
        }

        @Override
        public void run() {
            task.execute();
        }

    }

    static class Task {

        private ReentrantLock lock = new ReentrantLock();

        public void execute() {
            lock.lock();
            try {
                for (int i = 0; i < 3; i++) {
                    System.out.println(lock.toString());

                    // 查询当前线程 hold 住此锁的次数
                    System.out.println("\t holdCount: " + lock.getHoldCount());

                    // 查询正等待获取此锁的线程数
                    System.out.println("\t queuedLength: " + lock.getQueueLength());

                    // 是否为公平锁
                    System.out.println("\t isFair: " + lock.isFair());

                    // 是否被锁住
                    System.out.println("\t isLocked: " + lock.isLocked());

                    // 是否被当前线程持有锁
                    System.out.println("\t isHeldByCurrentThread: " + lock.isHeldByCurrentThread());

                    try {
                        Thread.sleep(500);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
            } finally {
                lock.unlock();
            }
        }

    }

}

输出结果：

1
2
3
4
5
6
7
8
9
10
11
12
13

java.util.concurrent.locks.ReentrantLock@64fcd88a[Locked by thread Thread-A]
	 holdCount: 1
	 queuedLength: 2
	 isFair: false
	 isLocked: true
	 isHeldByCurrentThread: true
java.util.concurrent.locks.ReentrantLock@64fcd88a[Locked by thread Thread-C]
	 holdCount: 1
	 queuedLength: 1
	 isFair: false
	 isLocked: true
	 isHeldByCurrentThread: true
// ...

tryLock 方法

与无条件获取锁相比，tryLock 有更完善的容错机制。

• tryLock() - 可轮询获取锁。如果成功，则返回 true；如果失败，则返回 false。也就是说，这个方法无论成败都会立即返回，获取不到锁（锁已被其他线程获取）时不会一直等待。
• tryLock(long, TimeUnit) - 可定时获取锁。和 tryLock() 类似，区别仅在于这个方法在获取不到锁时会等待一定的时间，在时间期限之内如果还获取不到锁，就返回 false。如果如果一开始拿到锁或者在等待期间内拿到了锁，则返回 true。

示例：ReentrantLock 的 tryLock() 操作

修改上个示例中的 execute() 方法

1
2
3
4
5
6
7
8
9
10
11
12
13

public void execute() {
    if (lock.tryLock()) {
        try {
            for (int i = 0; i < 3; i++) {
               // 略。..
            }
        } finally {
            lock.unlock();
        }
    } else {
        System.out.println(Thread.currentThread().getName() + " 获取锁失败");
    }
}

示例：ReentrantLock 的 tryLock(long, TimeUnit) 操作

修改上个示例中的 execute() 方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18

public void execute() {
    try {
        if (lock.tryLock(2, TimeUnit.SECONDS)) {
            try {
                for (int i = 0; i < 3; i++) {
                    // 略。..
                }
            } finally {
                lock.unlock();
            }
        } else {
            System.out.println(Thread.currentThread().getName() + " 获取锁失败");
        }
    } catch (InterruptedException e) {
        System.out.println(Thread.currentThread().getName() + " 获取锁超时");
        e.printStackTrace();
    }
}

lockInterruptibly 方法

• lockInterruptibly() - 可中断获取锁。可中断获取锁可以在获得锁的同时保持对中断的响应。可中断获取锁比其它获取锁的方式稍微复杂一些，需要两个 try-catch 块（如果在获取锁的操作中抛出了 InterruptedException ，那么可以使用标准的 try-finally 加锁模式）。
  • 举例来说：假设有两个线程同时通过 lock.lockInterruptibly() 获取某个锁时，若线程 A 获取到了锁，则线程 B 只能等待。若此时对线程 B 调用 threadB.interrupt() 方法能够中断线程 B 的等待过程。由于 lockInterruptibly() 的声明中抛出了异常，所以 lock.lockInterruptibly() 必须放在 try 块中或者在调用 lockInterruptibly() 的方法外声明抛出 InterruptedException。

:bell: 注意：当一个线程获取了锁之后，是不会被 interrupt() 方法中断的。单独调用 interrupt() 方法不能中断正在运行状态中的线程，只能中断阻塞状态中的线程。因此当通过 lockInterruptibly() 方法获取某个锁时，如果未获取到锁，只有在等待的状态下，才可以响应中断。

示例：ReentrantLock 的 lockInterruptibly() 操作

修改上个示例中的 execute() 方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14

public void execute() {
    try {
        lock.lockInterruptibly();

        for (int i = 0; i < 3; i++) {
            // 略。..
        }
    } catch (InterruptedException e) {
        System.out.println(Thread.currentThread().getName() + "被中断");
        e.printStackTrace();
    } finally {
        lock.unlock();
    }
}

newCondition 方法

newCondition() - 返回一个绑定到 Lock 对象上的 Condition 实例。Condition 的特性和具体方法请阅读下文 [Condition](#五 condition)。

ReentrantLock 的原理

ReentrantLock 的可见性

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

class X {
  private final Lock rtl =
  new ReentrantLock();
  int value;
  public void addOne() {
    // 获取锁
    rtl.lock();
    try {
      value+=1;
    } finally {
      // 保证锁能释放
      rtl.unlock();
    }
  }
}

ReentrantLock，内部持有一个 volatile 的成员变量 state，获取锁的时候，会读写 state 的值；解锁的时候，也会读写 state 的值（简化后的代码如下面所示）。也就是说，在执行 value+=1 之前，程序先读写了一次 volatile 变量 state，在执行 value+=1 之后，又读写了一次 volatile 变量 state。根据相关的 Happens-Before 规则：

1. 顺序性规则：对于线程 T1，value+=1 Happens-Before 释放锁的操作 unlock()；
2. volatile 变量规则：由于 state = 1 会先读取 state，所以线程 T1 的 unlock() 操作 Happens-Before 线程 T2 的 lock() 操作；
3. 传递性规则：线程 T1 的 value+=1 Happens-Before 线程 T2 的 lock() 操作。

ReentrantLock 的数据结构

阅读 ReentrantLock 的源码，可以发现它有一个核心字段：

1

private final Sync sync;

• sync - 内部抽象类 ReentrantLock.Sync 对象，Sync 继承自 AQS。它有两个子类：
• ReentrantLock.FairSync - 公平锁。
• ReentrantLock.NonfairSync - 非公平锁。

查看源码可以发现，ReentrantLock 实现 Lock 接口其实是调用 ReentrantLock.FairSync 或 ReentrantLock.NonfairSync 中各自的实现，这里不一一列举。

ReentrantLock 的获取锁和释放锁

ReentrantLock 获取锁和释放锁的接口，从表象看，是调用 ReentrantLock.FairSync 或 ReentrantLock.NonfairSync 中各自的实现；从本质上看，是基于 AQS 的实现。

仔细阅读源码很容易发现：

• void lock() 调用 Sync 的 lock() 方法。

• void lockInterruptibly() 直接调用 AQS 的 [获取可中断的独占锁](#获取可中断的独占锁） 方法 lockInterruptibly()。

• boolean tryLock() 调用 Sync 的 nonfairTryAcquire() 。

• boolean tryLock(long time, TimeUnit unit) 直接调用 AQS 的 [获取超时等待式的独占锁](#获取超时等待式的独占锁） 方法 tryAcquireNanos(int arg, long nanosTimeout)。

• void unlock() 直接调用 AQS 的 [释放独占锁](#释放独占锁） 方法 release(int arg) 。

直接调用 AQS 接口的方法就不再赘述了，其原理在 [AQS 的原理](#AQS 的原理) 中已经用很大篇幅进行过讲解。

nonfairTryAcquire 方法源码如下：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

// 公平锁和非公平锁都会用这个方法区尝试获取锁
final boolean nonfairTryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();
    if (c == 0) {
        if (compareAndSetState(0, acquires)) {
         // 如果同步状态为 0，将其设为 acquires，并设置当前线程为排它线程
            setExclusiveOwnerThread(current);
            return true;
        }
    }
    else if (current == getExclusiveOwnerThread()) {
        int nextc = c + acquires;
        if (nextc < 0) // overflow
            throw new Error("Maximum lock count exceeded");
        setState(nextc);
        return true;
    }
    return false;
}

处理流程很简单：

• 如果同步状态为 0，设置同步状态设为 acquires，并设置当前线程为排它线程，然后返回 true，获取锁成功。
• 如果同步状态不为 0 且当前线程为排它线程，设置同步状态为当前状态值+acquires 值，然后返回 true，获取锁成功。
• 否则，返回 false，获取锁失败。

公平锁和非公平锁

ReentrantLock 这个类有两个构造函数，一个是无参构造函数，一个是传入 fair 参数的构造函数。fair 参数代表的是锁的公平策略，如果传入 true 就表示需要构造一个公平锁，反之则表示要构造一个非公平锁。

锁都对应着一个等待队列，如果一个线程没有获得锁，就会进入等待队列，当有线程释放锁的时候，就需要从等待队列中唤醒一个等待的线程。如果是公平锁，唤醒的策略就是谁等待的时间长，就唤醒谁，很公平；如果是非公平锁，则不提供这个公平保证，有可能等待时间短的线程反而先被唤醒。

lock 方法在公平锁和非公平锁中的实现：

二者的区别仅在于申请非公平锁时，如果同步状态为 0，尝试将其设为 1，如果成功，直接将当前线程置为排它线程；否则和公平锁一样，调用 AQS 获取独占锁方法 acquire。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15

// 非公平锁实现
final void lock() {
    if (compareAndSetState(0, 1))
    // 如果同步状态为 0，将其设为 1，并设置当前线程为排它线程
        setExclusiveOwnerThread(Thread.currentThread());
    else
    // 调用 AQS 获取独占锁方法 acquire
        acquire(1);
}

// 公平锁实现
final void lock() {
    // 调用 AQS 获取独占锁方法 acquire
    acquire(1);
}

ReentrantReadWriteLock

ReadWriteLock 适用于读多写少的场景。

ReentrantReadWriteLock 类是 ReadWriteLock 接口的具体实现，它是一个可重入的读写锁。ReentrantReadWriteLock 维护了一对读写锁，将读写锁分开，有利于提高并发效率。

读写锁，并不是 Java 语言特有的，而是一个广为使用的通用技术，所有的读写锁都遵守以下三条基本原则：

• 允许多个线程同时读共享变量；
• 只允许一个线程写共享变量；
• 如果一个写线程正在执行写操作，此时禁止读线程读共享变量。

读写锁与互斥锁的一个重要区别就是读写锁允许多个线程同时读共享变量，而互斥锁是不允许的，这是读写锁在读多写少场景下性能优于互斥锁的关键。但读写锁的写操作是互斥的，当一个线程在写共享变量的时候，是不允许其他线程执行写操作和读操作。

ReentrantReadWriteLock 的特性

ReentrantReadWriteLock 的特性如下：

• ReentrantReadWriteLock 适用于读多写少的场景。如果是写多读少的场景，由于 ReentrantReadWriteLock 其内部实现比 ReentrantLock 复杂，性能可能反而要差一些。如果存在这样的问题，需要具体问题具体分析。由于 ReentrantReadWriteLock 的读写锁（ReadLock、WriteLock）都实现了 Lock 接口，所以要替换为 ReentrantLock 也较为容易。
• ReentrantReadWriteLock 实现了 ReadWriteLock 接口，支持了 ReentrantLock 所不具备的读写锁分离。ReentrantReadWriteLock 维护了一对读写锁（ReadLock、WriteLock）。将读写锁分开，有利于提高并发效率。ReentrantReadWriteLock 的加锁策略是：允许多个读操作并发执行，但每次只允许一个写操作。
• ReentrantReadWriteLock 为读写锁都提供了可重入的加锁语义。
• ReentrantReadWriteLock 支持公平锁和非公平锁（默认）两种模式。

ReadWriteLock 接口定义如下：

1
2
3
4

public interface ReadWriteLock {
    Lock readLock();
    Lock writeLock();
}

• readLock - 返回用于读操作的锁（ReadLock）。
• writeLock - 返回用于写操作的锁（WriteLock）。

在读写锁和写入锁之间的交互可以采用多种实现方式，ReadWriteLock 的一些可选实现包括：

• 释放优先 - 当一个写入操作释放写锁，并且队列中同时存在读线程和写线程，那么应该优先选择读线程、写线程，还是最先发出请求的线程？
• 读线程插队 - 如果锁是由读线程持有，但有写线程正在等待，那么新到达的读线程能否立即获得访问权，还是应该在写线程后面等待？如果允许读线程插队到写线程之前，那么将提高并发性，但可能造成线程饥饿问题。
• 重入性 - 读锁和写锁是否是可重入的？
• 降级 - 如果一个线程持有写入锁，那么它能否在不释放该锁的情况下获得读锁？这可能会使得写锁被降级为读锁，同时不允许其他写线程修改被保护的资源。
• 升级 - 读锁能否优先于其他正在等待的读线程和写线程而升级为一个写锁？在大多数的读写锁实现中并不支持升级，因为如果没有显式的升级操作，那么很容易造成死锁。

ReentrantReadWriteLock 的用法

前文了解了 ReentrantReadWriteLock 的特性，接下来，我们要讲述其具体用法。

ReentrantReadWriteLock 的构造方法

ReentrantReadWriteLock 和 ReentrantLock 一样，也有两个构造方法，且用法相似。

1
2

public ReentrantReadWriteLock() {}
public ReentrantReadWriteLock(boolean fair) {}

• ReentrantReadWriteLock() - 默认构造方法会初始化一个非公平锁（NonfairSync）。在非公平的锁中，线程获得锁的顺序是不确定的。写线程降级为读线程是可以的，但读线程升级为写线程是不可以的（这样会导致死锁）。
• ReentrantReadWriteLock(boolean) - new ReentrantLock(true) 会初始化一个公平锁（FairSync）。对于公平锁，等待时间最长的线程将优先获得锁。如果这个锁是读线程持有，则另一个线程请求写锁，那么其他读线程都不能获得读锁，直到写线程释放写锁。

ReentrantReadWriteLock 的使用实例

在 [ReentrantReadWriteLock 的特性](#reentrantreadwritelock-的特性） 中已经介绍过，ReentrantReadWriteLock 的读写锁（ReadLock、WriteLock) 都实现了 Lock 接口，所以其各自独立的使用方式与 ReentrantLock 一样，这里不再赘述。

ReentrantReadWriteLock 与 ReentrantLock 用法上的差异，主要在于读写锁的配合使用。本文以一个典型使用场景来进行讲解。

【示例】基于 ReadWriteLock 实现一个简单的泛型无界缓存

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55

/**
 * 简单的无界缓存实现
 * <p>
 * 使用 WeakHashMap 存储键值对。WeakHashMap 中存储的对象是弱引用，JVM GC 时会自动清除没有被引用的弱引用对象。
 */
static class UnboundedCache<K, V> {

    private final Map<K, V> cacheMap = new WeakHashMap<>();

    private final ReadWriteLock cacheLock = new ReentrantReadWriteLock();

    public V get(K key) {
        cacheLock.readLock().lock();
        V value;
        try {
            value = cacheMap.get(key);
            String log = String.format("%s 读数据 %s:%s", Thread.currentThread().getName(), key, value);
            System.out.println(log);
        } finally {
            cacheLock.readLock().unlock();
        }
        return value;
    }

    public V put(K key, V value) {
        cacheLock.writeLock().lock();
        try {
            cacheMap.put(key, value);
            String log = String.format("%s 写入数据 %s:%s", Thread.currentThread().getName(), key, value);
            System.out.println(log);
        } finally {
            cacheLock.writeLock().unlock();
        }
        return value;
    }

    public V remove(K key) {
        cacheLock.writeLock().lock();
        try {
            return cacheMap.remove(key);
        } finally {
            cacheLock.writeLock().unlock();
        }
    }

    public void clear() {
        cacheLock.writeLock().lock();
        try {
            this.cacheMap.clear();
        } finally {
            cacheLock.writeLock().unlock();
        }
    }

}

说明：

• 使用 WeakHashMap 而不是 HashMap 来存储键值对。WeakHashMap 中存储的对象是弱引用，JVM GC 时会自动清除没有被引用的弱引用对象。
• 向 Map 写数据前加写锁，写完后，释放写锁。
• 向 Map 读数据前加读锁，读完后，释放读锁。

测试其线程安全性：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31

/**
 * @author <a href="mailto:forbreak@163.com">Zhang Peng</a>
 * @since 2020-01-01
 */
public class ReentrantReadWriteLockDemo {

    static UnboundedCache<Integer, Integer> cache = new UnboundedCache<>();

    public static void main(String[] args) {
        ExecutorService executorService = Executors.newCachedThreadPool();
        for (int i = 0; i < 20; i++) {
            executorService.execute(new MyThread());
            cache.get(0);
        }
        executorService.shutdown();
    }

    /** 线程任务每次向缓存中写入 3 个随机值，key 固定 */
    static class MyThread implements Runnable {

        @Override
        public void run() {
            Random random = new Random();
            for (int i = 0; i < 3; i++) {
                cache.put(i, random.nextInt(100));
            }
        }

    }

}

说明：示例中，通过线程池启动 20 个并发任务。任务每次向缓存中写入 3 个随机值，key 固定；然后主线程每次固定读取缓存中第一个 key 的值。

输出结果：

1
2
3
4
5
6
7
8
9
10
11
12
13

main 读数据 0:null
pool-1-thread-1 写入数据 0:16
pool-1-thread-1 写入数据 1:58
pool-1-thread-1 写入数据 2:50
main 读数据 0:16
pool-1-thread-1 写入数据 0:85
pool-1-thread-1 写入数据 1:76
pool-1-thread-1 写入数据 2:46
pool-1-thread-2 写入数据 0:21
pool-1-thread-2 写入数据 1:41
pool-1-thread-2 写入数据 2:63
main 读数据 0:21
// ...

ReentrantReadWriteLock 的原理

前面了解了 ReentrantLock 的原理，理解 ReentrantReadWriteLock 就容易多了。

ReentrantReadWriteLock 的数据结构

阅读 ReentrantReadWriteLock 的源码，可以发现它有三个核心字段：

1
2
3
4
5
6
7
8
9

/** Inner class providing readlock */
private final ReentrantReadWriteLock.ReadLock readerLock;
/** Inner class providing writelock */
private final ReentrantReadWriteLock.WriteLock writerLock;
/** Performs all synchronization mechanics */
final Sync sync;

public ReentrantReadWriteLock.WriteLock writeLock() { return writerLock; }
public ReentrantReadWriteLock.ReadLock  readLock()  { return readerLock; }

• sync - 内部类 ReentrantReadWriteLock.Sync 对象。与 ReentrantLock 类似，它有两个子类：ReentrantReadWriteLock.FairSync 和 ReentrantReadWriteLock.NonfairSync ，分别表示公平锁和非公平锁的实现。
• readerLock - 内部类 ReentrantReadWriteLock.ReadLock 对象，这是一把读锁。
• writerLock - 内部类 ReentrantReadWriteLock.WriteLock 对象，这是一把写锁。

ReentrantReadWriteLock 的获取锁和释放锁

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

public static class ReadLock implements Lock, java.io.Serializable {

    // 调用 AQS 获取共享锁方法
    public void lock() {
        sync.acquireShared(1);
    }

    // 调用 AQS 释放共享锁方法
    public void unlock() {
        sync.releaseShared(1);
    }
}

public static class WriteLock implements Lock, java.io.Serializable {

    // 调用 AQS 获取独占锁方法
    public void lock() {
        sync.acquire(1);
    }

    // 调用 AQS 释放独占锁方法
    public void unlock() {
        sync.release(1);
    }
}

StampedLock

ReadWriteLock 支持两种模式：一种是读锁，一种是写锁。而 StampedLock 支持三种模式，分别是：写锁、悲观读锁和乐观读。其中，写锁、悲观读锁的语义和 ReadWriteLock 的写锁、读锁的语义非常类似，允许多个线程同时获取悲观读锁，但是只允许一个线程获取写锁，写锁和悲观读锁是互斥的。不同的是：StampedLock 里的写锁和悲观读锁加锁成功之后，都会返回一个 stamp；然后解锁的时候，需要传入这个 stamp。

注意这里，用的是“乐观读”这个词，而不是“乐观读锁”，是要提醒你，乐观读这个操作是无锁的，所以相比较 ReadWriteLock 的读锁，乐观读的性能更好一些。

StampedLock 的性能之所以比 ReadWriteLock 还要好，其关键是 StampedLock 支持乐观读的方式。

• ReadWriteLock 支持多个线程同时读，但是当多个线程同时读的时候，所有的写操作会被阻塞；
• 而 StampedLock 提供的乐观读，是允许一个线程获取写锁的，也就是说不是所有的写操作都被阻塞。

对于读多写少的场景 StampedLock 性能很好，简单的应用场景基本上可以替代 ReadWriteLock，但是，StampedLock 的功能仅仅是 ReadWriteLock 的子集，在使用的时候，还是有几个地方需要注意一下。

• StampedLock 不支持重入
• StampedLock 的悲观读锁、写锁都不支持条件变量。
• 如果线程阻塞在 StampedLock 的 readLock() 或者 writeLock() 上时，此时调用该阻塞线程的 interrupt() 方法，会导致 CPU 飙升。**使用 StampedLock 一定不要调用中断操作，如果需要支持中断功能，一定使用可中断的悲观读锁 readLockInterruptibly() 和写锁 writeLockInterruptibly()**。

【示例】StampedLock 阻塞时，调用 interrupt() 导致 CPU 飙升

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21

final StampedLock lock = new StampedLock();
Thread T1 = new Thread(() -> {
    // 获取写锁
    lock.writeLock();
    // 永远阻塞在此处，不释放写锁
    LockSupport.park();
});
T1.start();
// 保证 T1 获取写锁
Thread.sleep(100);
Thread T2 = new Thread(() ->
    // 阻塞在悲观读锁
    lock.readLock()
);
T2.start();
// 保证 T2 阻塞在读锁
Thread.sleep(100);
// 中断线程 T2
// 会导致线程 T2 所在 CPU 飙升
T2.interrupt();
T2.join();

【示例】StampedLock 读模板：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

final StampedLock sl = new StampedLock();

// 乐观读
long stamp = sl.tryOptimisticRead();
// 读入方法局部变量
// ......
// 校验 stamp
if (!sl.validate(stamp)) {
    // 升级为悲观读锁
    stamp = sl.readLock();
    try {
        // 读入方法局部变量
        // .....
    } finally {
        // 释放悲观读锁
        sl.unlockRead(stamp);
    }
}
// 使用方法局部变量执行业务操作
// ......

【示例】StampedLock 写模板：

1
2
3
4
5
6
7

long stamp = sl.writeLock();
try {
  // 写共享变量
  ......
} finally {
  sl.unlockWrite(stamp);
}

参考资料

• 《Java 并发编程实战》
• 《Java 并发编程的艺术》
• 极客时间教程 - Java 并发编程实战
• Java 并发编程：Lock
• 深入学习 java 同步器 AQS
• AbstractQueuedSynchronizer 框架
• Java 中的锁分类

Java 并发之无锁

并发安全需要保证几个基本特性：

• 可见性 - 是一个线程修改了某个共享变量，其状态能够立即被其他线程知晓，通常被解释为将线程本地状态反映到主内存上，volatile 就是负责保证可见性的。
• 有序性 - 是保证线程内串行语义，避免指令重排等。
• 原子性 - 简单说就是相关操作不会中途被其他线程干扰，一般通过互斥机制（加锁：sychronized、Lock）实现。

互斥同步是最常见的原子性保障手段。互斥同步最主要的问题是线程阻塞和唤醒所带来的性能问题。因此，互斥同步也被称为阻塞同步。互斥同步属于一种悲观的并发策略，总是认为只要不去做正确的同步措施，那就肯定会出现问题。无论共享数据是否真的会出现竞争，它都要进行加锁（这里讨论的是概念模型，实际上虚拟机会优化掉很大一部分不必要的加锁）、用户态核心态转换、维护锁计数器和检查是否有被阻塞的线程需要唤醒等操作。

解决并发安全问题，还可以采用无锁方案。无锁方案相对互斥锁方案，最大的好处就是性能。互斥锁方案为了保证互斥性，需要执行加锁、解锁操作，而加锁、解锁操作本身就消耗性能；同时拿不到锁的线程还会进入阻塞状态，进而触发线程切换，线程切换对性能的消耗也很大。 相比之下，无锁方案则完全没有加锁、解锁的性能消耗，同时还能保证互斥性。

Java 中的无锁技术有：

• CAS
• 原子类
• ThreadLocal
• Copy-on-Write
• 不变模式

CAS

CAS 的要点

CAS（Compare and Swap），字面意思为比较并交换。

CAS 涉及三个操作数：

• V：需要读写的内存位置
• A：进行比较的值
• B：拟写入的新值

当且仅当 V 的值等于 A 时，才会通过原子方式用新值 B 来更新 A 的值，否则什么都不做。

CAS 实际是乐观锁的一种实现方式，因此，CAS 只适用于线程冲突较少的情况。

CAS 的应用

CAS 的典型应用场景是：

• 原子类
• 自旋锁

原子类

原子类是 CAS 在 Java 中最典型的应用。

我们先来看一个常见的代码片段。

1
2
3

if(a==b) {
    a++;
}

如果 a++ 执行前， a 的值被修改了怎么办？还能得到预期值吗？出现该问题的原因是在并发环境下，以上代码片段不是原子操作，随时可能被其他线程所篡改。

解决这种问题的最经典方式是应用原子类的 incrementAndGet 方法。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

public class AtomicIntegerDemo {

    public static void main(String[] args) throws InterruptedException {
        ExecutorService executorService = Executors.newFixedThreadPool(3);
        final AtomicInteger count = new AtomicInteger(0);
        for (int i = 0; i < 10; i++) {
            executorService.execute(new Runnable() {
                @Override
                public void run() {
                    count.incrementAndGet();
                }
            });
        }

        executorService.shutdown();
        executorService.awaitTermination(3, TimeUnit.SECONDS);
        System.out.println("Final Count is : " + count.get());
    }

}

J.U.C 包中提供了 AtomicBoolean、AtomicInteger、AtomicLong 分别针对 Boolean、Integer、Long 执行原子操作，操作和上面的示例大体相似，不做赘述。

自旋锁

利用原子类（本质上是 CAS），可以实现自旋锁。

所谓自旋锁，是指线程反复检查锁变量是否可用，直到成功为止。由于线程在这一过程中保持执行，因此是一种忙等待。一旦获取了自旋锁，线程会一直保持该锁，直至显式释放自旋锁。

示例：非线程安全示例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25

public class AtomicReferenceDemo {

    private static int ticket = 10;

    public static void main(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(3);
        for (int i = 0; i < 5; i++) {
            executorService.execute(new MyThread());
        }
        executorService.shutdown();
    }

    static class MyThread implements Runnable {

        @Override
        public void run() {
            while (ticket > 0) {
                System.out.println(Thread.currentThread().getName() + " 卖出了第 " + ticket + " 张票");
                ticket--;
            }
        }

    }

}

输出结果：

1
2
3
4
5
6
7
8
9
10
11
12

pool-1-thread-2 卖出了第 10 张票
pool-1-thread-1 卖出了第 10 张票
pool-1-thread-3 卖出了第 10 张票
pool-1-thread-1 卖出了第 8 张票
pool-1-thread-2 卖出了第 9 张票
pool-1-thread-1 卖出了第 6 张票
pool-1-thread-3 卖出了第 7 张票
pool-1-thread-1 卖出了第 4 张票
pool-1-thread-2 卖出了第 5 张票
pool-1-thread-1 卖出了第 2 张票
pool-1-thread-3 卖出了第 3 张票
pool-1-thread-2 卖出了第 1 张票

很明显，出现了重复售票的情况。

【示例】使用自旋锁来保证线程安全

可以通过自旋锁这种非阻塞同步来保证线程安全，下面使用 AtomicReference 来实现一个自旋锁。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56

public class AtomicReferenceDemo2 {

    private static int ticket = 10;

    public static void main(String[] args) {
        threadSafeDemo();
    }

    private static void threadSafeDemo() {
        SpinLock lock = new SpinLock();
        ExecutorService executorService = Executors.newFixedThreadPool(3);
        for (int i = 0; i < 5; i++) {
            executorService.execute(new MyThread(lock));
        }
        executorService.shutdown();
    }

    static class SpinLock {

        private AtomicReference<Thread> atomicReference = new AtomicReference<>();

        public void lock() {
            Thread current = Thread.currentThread();
            while (!atomicReference.compareAndSet(null, current)) {}
        }

        public void unlock() {
            Thread current = Thread.currentThread();
            atomicReference.compareAndSet(current, null);
        }

    }

    static class MyThread implements Runnable {

        private SpinLock lock;

        public MyThread(SpinLock lock) {
            this.lock = lock;
        }

        @Override
        public void run() {
            while (ticket > 0) {
                lock.lock();
                if (ticket > 0) {
                    System.out.println(Thread.currentThread().getName() + " 卖出了第 " + ticket + " 张票");
                    ticket--;
                }
                lock.unlock();
            }
        }

    }

}

输出结果：

1
2
3
4
5
6
7
8
9
10

pool-1-thread-2 卖出了第 10 张票
pool-1-thread-1 卖出了第 9 张票
pool-1-thread-3 卖出了第 8 张票
pool-1-thread-2 卖出了第 7 张票
pool-1-thread-3 卖出了第 6 张票
pool-1-thread-1 卖出了第 5 张票
pool-1-thread-2 卖出了第 4 张票
pool-1-thread-1 卖出了第 3 张票
pool-1-thread-3 卖出了第 2 张票
pool-1-thread-1 卖出了第 1 张票

CAS 的原理

在 Java 中，主要利用 Unsafe 这个类实现 CAS。

Unsafe 类位于 sun.misc 包下，是一个提供低级别、不安全操作的类。由于其强大的功能和潜在的危险性，它通常用于 JVM 内部或一些需要极高性能和底层访问的库中，而不推荐普通开发者在应用程序中使用。

Unsafe 类提供了 compareAndSwapObject、compareAndSwapInt、compareAndSwapLong方法来实现的对 Object、int、long 类型的 CAS 操作：

1
2
3
4
5
6
7
8
9
10
11
12
13
14

/**
 * 以原子方式更新对象字段的值。
 */
boolean compareAndSwapObject(Object o, long offset, Object expected, Object x);

/**
 * 以原子方式更新 int 类型的对象字段的值。
 */
boolean compareAndSwapInt(Object o, long offset, int expected, int x);

/**
 * 以原子方式更新 long 类型的对象字段的值。
 */
boolean compareAndSwapLong(Object o, long offset, long expected, long x);

Unsafe 类中的 CAS 方法是 native 方法。native 关键字表明这些方法是用本地代码（通常是 C 或 C++）实现的，而不是用 Java 实现的。这些方法直接调用底层的、具有原子性的 CPU 指令来实现。

由于 CAS 操作可能会因为并发冲突而失败，因此通常会伴随着自旋，而所谓自旋，其实就是循环尝试。

Unsafe#getAndAddInt 源码：

1
2
3
4
5
6
7
8
9
10

// 原子地获取并增加整数值
public final int getAndAddInt(Object o, long offset, int delta) {
    int v;
    do {
        // 以 volatile 方式获取对象 o 在内存偏移量 offset 处的整数值
        v = getIntVolatile(o, offset);
    } while (!compareAndSwapInt(o, offset, v, v + delta));
    // 返回旧值
    return v;
}

CAS 的问题

一般情况下，CAS 比锁性能更高。因为 CAS 是一种非阻塞算法，所以其避免了线程阻塞和唤醒的等待时间。但是，事物总会有利有弊，CAS 也存在三大问题：

• ABA 问题
• 循环时间长开销大
• 只能保证一个共享变量的原子性

ABA 问题

如果一个变量初次读取的时候是 A 值，它的值被改成了 B，后来又被改回为 A，那 CAS 操作就会误认为它从来没有被改变过，这就是 ABA 问题。

ABA 问题的解决思路是在变量前面追加上版本号或者时间戳。J.U.C 包提供了一个带有标记的原子引用类 AtomicStampedReference 来解决这个问题，它可以通过控制变量值的版本来保证 CAS 的正确性。大部分情况下 ABA 问题不会影响程序并发的正确性，如果需要解决 ABA 问题，改用传统的互斥同步可能会比原子类更高效。

循环时间长开销大

自旋 CAS （不断尝试，直到成功为止）如果长时间不成功，会给 CPU 带来非常大的执行开销。

如果 JVM 能支持处理器提供的 pause 指令那么效率会有一定的提升，pause 指令有两个作用：

• 它可以延迟流水线执行指令（de-pipeline）, 使 CPU 不会消耗过多的执行资源，延迟的时间取决于具体实现的版本，在一些处理器上延迟时间是零。
• 它可以避免在退出循环的时候因内存顺序冲突（memory order violation）而引起 CPU 流水线被清空（CPU pipeline flush），从而提高 CPU 的执行效率。

比较花费 CPU 资源，即使没有任何用也会做一些无用功。

只能保证一个共享变量的原子性

当对一个共享变量执行操作时，我们可以使用循环 CAS 的方式来保证原子操作，但是对多个共享变量操作时，循环 CAS 就无法保证操作的原子性，这个时候就可以用锁。

或者有一个取巧的办法，就是把多个共享变量合并成一个共享变量来操作。比如有两个共享变量 i ＝ 2, j = a，合并一下 ij=2a，然后用 CAS 来操作 ij。从 Java 1.5 开始 JDK 提供了 AtomicReference 类来保证引用对象之间的原子性，你可以把多个变量放在一个对象里来进行 CAS 操作。

原子类

原子类简介

原子性是确保并发安全三大特性之一。为了兼顾原子性以及锁带来的性能问题，Java 引入了 CAS （主要体现在 Unsafe 类）来实现非阻塞同步（也叫乐观锁），CAS 底层基于 CPU 指令（硬件支持）支持，具有原子性。并基于 CAS ，提供了一套原子工具类。

原子类比锁的粒度更细，更轻量级，并且对于在多处理器系统上实现高性能的并发代码来说是非常关键的。原子变量将发生竞争的范围缩小到单个变量上。

原子类相当于一种泛化的 volatile 变量，能够支持原子的、有条件的读/改/写操作。

原子类可以分为 5 个类别，这 5 个类别提供的方法基本上是相似的：

• 基本数据类型
  • AtomicBoolean - 布尔类型原子类
  • AtomicInteger - 整型原子类
  • AtomicLong - 长整型原子类
• 引用数据类型
  • AtomicReference - 引用类型原子类
  • AtomicMarkableReference - 带有标记位的引用类型原子类
  • AtomicStampedReference - 带有版本号的引用类型原子类
• 数组数据类型
  • AtomicIntegerArray - 整形数组原子类
  • AtomicLongArray - 长整型数组原子类
  • AtomicReferenceArray - 引用类型数组原子类
• 属性更新器类型
  • AtomicIntegerFieldUpdater - 整型字段的原子更新器
  • AtomicLongFieldUpdater - 长整型字段的原子更新器
  • AtomicReferenceFieldUpdater - 原子更新引用类型里的字段
• 累加器
  • DoubleAdder - 浮点型原子累加器
  • LongAdder - 长整型原子累加器
  • DoubleAccumulator - 更复杂的浮点型原子累加器
  • LongAccumulator - 更复杂的长整型原子累加器

原子类之基本数据类型

基本数据类型原子类针对 Java 基本类型提供原子操作。

• AtomicBoolean - 布尔类型原子类
• AtomicInteger - 整型原子类
• AtomicLong - 长整型原子类

以上类都支持 CAS（compare-and-swap）技术，此外，AtomicInteger、AtomicLong 还支持算术运算。

:bulb: 提示：

虽然 Java 只提供了 AtomicBoolean 、AtomicInteger、AtomicLong，但是可以模拟其他基本类型的原子变量。要想模拟其他基本类型的原子变量，可以将 short 或 byte 等类型与 int 类型进行转换，以及使用 Float.floatToIntBits 、Double.doubleToLongBits 来转换浮点数。

由于 AtomicBoolean、AtomicInteger、AtomicLong 实现方式、使用方式都相近，所以本文仅针对 AtomicInteger 进行介绍。