【Java并发工具类】Java并发容器

前言

Java并发包有很大一部分都是关于并发容器的。Java在5.0版本之前线程安全的容器称之为同步容器。同步容器实现线程安全的方式：是将每个公有方法都使用synchronized修饰，保证每次只有一个线程能访问容器的状态。但是这样的串行度太高，将严重降低并发性，当多个线程竞争容器的锁时，吞吐量将严重降低。因此，在Java 5.0版本时提供了性能更高的容器来改进之前的同步容器，我们称其为并发容器。

下面我们先来介绍Java 5.0之前的同步容器，然后再来介绍Java 5.0之后的并发容器。

Java 5.0之前的同步容器

目前，Java中的容器主要可分为四大类，分别为List、Map、Set、Queue（Queue是Java5.0添加的新的容器类型），但是并不是所有的Java容器都是线程安全的。例如，我们常用的ArrayList和HashMap就不是线程安全的。线程安全的类为Vector、Stack和HashTable。

如何将非线程安全的类变为线程安全的类？
非线程安全的容器类可以由Collections类提供的Collections.synchronizedXxx()工厂方法将其包装为线程安全的类。

// 分别将ArrayList、HashMap和HashSet包装成线程安全的List 、Map和Set
List list = Collections.synchronizedList(new ArrayList());
Set set = Collections.synchronizedSet(new HashSet());
Map map = Collections.synchronizedMap(new HashMap());

这些类实现线程安全的方式是：将它们的状态封装起来，并对每个公有方法进行同步，使得每次只有一个线程能访问容器的状态。以ArrayList为例，可以使用如下的代码来理解如何将非线程安全的容器包装为线程安全的容器。

// 包装 ArrayList 
SafeArrayList<T>{
    List<T> c = new ArrayList<>();
    // 控制访问路径，使用synchronized修饰保证线程互斥访问
    synchronized T get(int idx){
        return c.get(idx);
    }
    synchronized void add(int idx, T t) {
        c.add(idx, t);
    }
    synchronized boolean addIfNotExist(T t){
        if(!c.contains(t)) {
            c.add(t);
            return true;
        }
        return false;
    }
}

被包装出来的线程安全的类，都是基于synchronized同步关键字实现，于是被成为同步容器。而原本的线程安全的容器类Vector等，同样也是基于synchronized关键字实现的。

同步容器在复合操作中的问题

同步容器类都是线程安全的，但是复合操作往往都会包含竞态条件问题。这时就需要额外的客户端加锁来保证复合操作的原子性。

在下例\(^{[2]}\)中，定义了两个方法getLast()和deleteLast()，它们都会执行“先检查后执行再运行”操作。每个方法首先都获得数组的大小，然后通过结果来获取或删除最后一个元素。

public class UnsafeVectorHelpers {
    public static Object getLast(Vector list) {
        int lastIndex = list.size() - 1;
        return list.get(lastIndex);
    }

    public static void deleteLast(Vector list) {
        int lastIndex = list.size() - 1;
        list.remove(lastIndex);
    }
}

如果线程同时调用相同的方法，这将不会产生什么问题。但是从调用者方向看，这将导致非常严重的后果。如果线程A在包含10个元素的Vector上调用getLast，同时线程B在同一个Vector上调用deleteLast，这些操作的交替若如下所示，那么getLast将抛出ArrayIndexOutOfBoundsException异常。

线程A在调用size()与getLast()这两个操作之间，Vector变小了，因此在调用size时得到的索引值将不再有效。

于是我们便需要在客户端加锁实现新操作的原子性。那么就需要考虑对哪个锁对象进行加锁。
同步容器类通过加锁自身（this）来保护它的每个方法，于是在这里我们锁住list对象便可以保证getLast()和deleteLast()成为原子性操作。

public class SafeVectorHelpers {
    public static Object getLast(Vector list) {
        synchronized (list) {
            int lastIndex = list.size() - 1;
            return list.get(lastIndex);
        }
    }

    public static void deleteLast(Vector list) {
        synchronized (list) {
            int lastIndex = list.size() - 1;
            list.remove(lastIndex);
        }
    }
}

在对Vector中元素进行迭代\(^{[2]}\)时，调用size()和相应的get()之间Vector的大小可能发生变化的情况也会出现。

for(int i=0; i<vector.size(); i++){
    doSomething(vector.get(i));
}

与getLast()一样，如果在对Vector进行迭代时，另一个线程删除了一个元素，并且删除和访问这两个操作交替执行，那么上面的方法将抛出ArrayIndexOutOfBoundsException异常。
同样，我们可以通过在客户端加锁来防止其他线程在迭代期间修改Vector。

synchronized(vector){
    for(int i=0; i<vector.size(); i++){
        doSomething(vector.get(i));
    }
}

有得必有失，以上代码将会导致其他线程在迭代期间无法访问vector，因此也降低了并发性。

迭代器与ConcurrentModificationException

无论是使用for循环迭代，还是使用Java 5.0引入的for-each循环语法，对容器类进行迭代的标准方式都是使用Iterator。同样，如果在使用迭代器访问容器期间，有线程并发地修改容器的大小，也是需要对迭代操作进行加锁，即如下\({^{[1]}}\)。

List list = Collections.synchronizedList(new ArrayList());
synchronized (list) {??
    Iterator i = list.iterator();
    while (i.hasNext())
        foo(i.next());
}? ?

在设计同步容器类的迭代器时没有考虑到并发修改的问题，当出现如上情况时，它们表现出来的行为是“及时失败”（fail-fast）的。当它们发现容器在迭代过程中被修改时，就会立即抛出一个ConcurrentModificationException异常。

这种fail-fast的迭代器并不是一种完备的处理机制，而只是“善意地”捕获并发错误，因此只能作为并发问题的预警指示器。这种机制的实现方式是：使用一个计数器modCount记录容器大小改变的次数，在进行迭代期间，如果该计数器值与刚进入迭代时不一致，那么hasNext()或next()将抛出ConcurrentModificationException异常。

但是，对计数器的值的检查时是没有在同步情况下进行的，因此可能会看到失效的计数值，导致迭代器没有意识到容器已经发生了修改。这是一种设计上的权衡，从而降低并发修改操作的检测代码对程序性能带来的影响。

更多的时候，我们是不希望在迭代期间对容器加锁。如果容器规模很大，在加锁迭代后，那么在迭代期间其他线程都不能访问该容器。这将降低程序的可伸缩性以及引起激烈的锁竞争降低吞吐量和CPU利用率。
一种替代加锁迭代的方法为“克隆”容器，并在副本上迭代。副本是线程封闭的，自然也就是安全的。但是克隆的过程也需要对容器加锁，且也存在一定的开销，需考虑使用。

隐藏的迭代器

容器的hashCode()和equals()等方法也会间接地执行迭代操作，当容器作为另一个容器的元素或键值时，就会出现这种情况。同样，containsAll()、removeAll()和retainAll()等方法，以及把容器作为参数的构造函数，都会对容器进行迭代。所有这些间接迭代操作都可能抛出ConcurrentModificationException异常。

Java 5.0的并发容器

在Java 5.0版本时提供了性能更高的容器来改进之前的同步容器，我们称之为并发容器。并发容器虽然多，但是总结下来依旧为四大类：List、Map、Set、Queue。

List

CopyOnWriteArrayList是用于替代同步List的并发容器，在迭代期间不需要对容器进行加锁或复制。写时复制（CopyOnWrite）的线程安全性体现在，只要正确地发布一个事实不可变的对象，那么在访问该对象时就不再需要进一步的同步。而在每次进行写操作时，便会创建一个副本出来，从而实现可变性。“写时复制”容器返回的迭代器不会抛出ConcurrentModificationException，因为迭代器在迭代过程中，如果对象会被修改则会创建一个副本被修改，被迭代的对象依旧是原来的。

CopyOnWriteArrayList仅适用于写操作非常少的场景，而且能够容忍短暂的不一致。CopyOnWriteArrayList迭代器是只读的，不支持增删改。因为迭代器遍历的仅仅是一个快照，而对快照进行增删改是没有意义的。

Map

ConcurrentHashMap和ConcurrentSkipListMap的区别为：ConcurrentHashMap的key是无序的，而ConcurrentSkipListMap的key是有序的。使用这两者时，它们的key和value都不能为空，否则会抛出NullPointerException异常。

Map有关实现类对于key和value的要求：

与HashMap一样，ConcurrentHashMap也是一个基于散列的Map，它使用分段锁实现了更大程度的共享。任意数量的读取线程可以并发地访问Map，执行读取的线程和执行写入的线程可以并发地访问Map，并且一定数量的写入线程可以并发地修改Map。ConcurrentHashMap在并发环境下可以实现更高的吞吐量，而在单线程环境中只损失非常小的性能。

ConcurrentHashMap返回的迭代器也不会抛出ConcurrentModificationException，因此不需要在迭代期间对容器加锁。ConcurrentHashMap返回的迭代器具有弱一致性（Weakly Consistent）,而并非fail-fast的。弱一致性的迭代器可以容忍并发的修改，当创建迭代器会遍历已有的元素，并可以（但是不保证）在迭代器被构建后将修改操作反映给容器。

ConcurrentHahsMap是对Map进行分段加锁，没有实现独占。所有需要独占访问功能的，应该使用其他并发容器。

ConcurrentSkipListMap里面的SkipList本身就是一种数据结构，中文翻译为“跳表”。跳表插入、删除、查询操作平均时间复杂度为O(log n)。返回的迭代器也是弱一致性的，也不会抛出ConcurrentModificationException。

Set

Set接口下，两个并发容器是CopyOnWriteArraySet和ConcurrentSkipListSet，可参考CopyOnWriteArrayList和ConcurrentSkipListMap理解。

Queue

Java并发包中Queue下的并发容器是最复杂的，可以从下面两个维度来分类：

1. 阻塞和非阻塞

   阻塞队列是指当队列已满时，入队操作阻塞；当队列为空时，出对操作阻塞。

2. 单端和双端

   单端队列指的是只能从队尾入队，队首出队；双端指的是队首队尾皆可出队。

在Java并发包中，阻塞队列都有Blocking标识，单端队列是用Queue标识，而双端队列是Deque标识。以上两个维度可组合，于是分为四类并发容器：单端阻塞队列、双端阻塞队列、单端非阻塞队列、双端非阻塞队列。

在使用队列时，需要格外注意队列是否为有界队列（内部的队列是否容量有限），无界队列在数据量大时，会导致OOM即内存溢出。
在有Queue的具体实现的并发容器中，只有ArrayBlockingQueue和LinkedBlockingQueue是支持有界的，其余都是无界队列。

小结

这篇文章从宏观层面介绍了Java并发包中的并发工具类，对每个容器类仅做了简单介绍，后续将附文介绍每一个容器类。

参考：
[1]极客时间专栏王宝令《Java并发编程实战》
[2]Brian Goetz.Tim Peierls. et al.Java并发编程实战[M].北京:机械工业出版社,2016