Java容器笔记（四）：认识并发容器类

对于线程安全和并发：线程安全并不一定适合并发（性能还要好），要根据应用场景选用最合适的容器类。

一、历史

• JDK1.0
  同步容器类：Vector和Hashtable

• **JDK1.2 **
  Collections工具类中的synchronizedXxx方法：
  返回指定集合对象对应的同步对象。synchronizedXxx方法本质是对相应容器的包装。

synchronized_methods.png

• **JDK1.5及之后 **
  java.util.concurrent中的多种并发容器
  并发容器包注重以下特性：

（1）根据具体场景进行设计，尽量避免使用锁，提高容器的并发访问性。
（2）并发容器定义了一些线程安全的复合操作。
（3）并发容器在迭代时，可以不封闭在synchronized中。但是未必每次看到的都是"最新的、当前的"数据。如果说将迭代操作包装在synchronized中，可以达到"串行"的并发安全性，那么并发容器的迭代达到了"脏读"。

可以通过下图简单了解concurrent中关于容器类的接口和类：

Concurrent.png

二、并发容器

2.1 接口

• ConcurrentMap
  该接口定义Map的原子操作：putIfAbsent、remove、replace。

A Map providing thread safety and atomicity guarantees.
Memory consistency effects:
As with other concurrent collections, 
actions in a thread prior to placing an object into a ConcurrentMap as a key or value 
happen-before
actions subsequent to the access or removal of that object from the ConcurrentMap in another thread.

• BlockingQueue
  阻塞队列，不允许null值；
  取元素时，如果队列为空则等待；存元素时，如果没有空间则等待；

A Queue that additionally supports operations:
wait for the queue to become non-empty when retrieving an element, 
and wait for space to become available in the queue when storing an element.

阻塞队列的方法有四种形式--当操作不能立即得到满足，但可能在未来某一时刻被满足的时候，有四种不同的方式来处理：
a.抛出异常
b.返回特殊的值（null或false，取决与具体的操作）
c.无期限地阻塞当前线程，直到该操作成功
d.仅在指定的最大时长内阻塞，过后还不成功就放弃

BlockingQueue methods come in four forms, with different ways of handling operations 
that cannot be satisfied immediately, but may be satisfied at some point in the future:
 * one throws an exception, 
 * the second returns a special value (either null or  false, depending on the operation), 
 * the third blocks the current thread indefinitely until the operation can succeed,
 * and the fourth blocks for only a given maximum time limit before giving up. 

阻塞队列主要是为生产者-消费者模型设计，但也支持Collection接口，所以它也可以使用remove(x)来移除任意一个元素，但这种操作通常效率不高，偶尔使用时可以的，比如当一个队列消息被取消的时候。

BlockingQueue implementations are designed to be usedprimarily for producer-consumer queues, 
but additionally support the java.util.Collection interface.  
So, for example, it is possible to remove an arbitrary element from a queue using remove(x). 
However, such operations are in general not performed very efficiently, 
and are intended for only occasional use, 
such as when a queued message is cancelled.

阻塞队列的实现类是线程安全的。所有的队列方法可以原子地达到它们的效果：使用内部锁或者其他形式的并发控制。但是，对于批量集合操作，如addAll、containsAll、retainAll、removeAll并不是原子性的（除非另有规定）。所以，像addAll(c)这个操作可能会抛出异常而失败（因为并没有将c中的所有元素都加进来）

BlockingQueue implementations are thread-safe.  
All queuing methods achieve their effects atomically 
using internal locks or other forms of concurrency control. 
However, the bulk Collection operations addAll,containsAll,  retainAll and removeAll 
are not necessarily performed atomically unless specified otherwise in an implementation. 
So it is possible, for example, for addAll(c) to fail (throwing an exception) 
after adding only some of the elements in  c.

• BlockingDeque
  支持阻塞操作的Deque，与BlockingQueue类似，只是双向队列。

2.2 实现类

通过上面的图可以知道，concurrent包中的并发容器主要可以分为：

• CopyOnWrite容器：CopyOnWriteArrayList、CopyOnWriteArraySet
• CocurrentMap的实现类：ConcurrentHashMap、ConcurrentSkipListMap
• 阻塞队列的实现类：见上图
• 其他：ConcurrentLinkedQueue、ConcurrentLikedDeque、ConcurrentSkipListSet

2.2.1 CopyOnWrite容器

该部分内容（概念、应用场景、缺点）摘抄自JAVA中的COPYONWRITE容器</br>
Copy-On-Write简称COW，是一种用于程序设计中的优化策略。其基本思路是，从一开始大家都在共享同一个内容，当某个人想要修改这个内容的时候，才会真正把内容Copy出去形成一个新的内容然后再改，这是一种延时懒惰策略。

• 概念
  CopyOnWrite容器即写时复制的容器。通俗的理解是当我们往一个容器添加元素的时候，不直接往当前容器添加，而是先将当前容器进行Copy，复制出一个新的容器，然后新的容器里添加元素，添加完元素之后，再将原容器的引用指向新的容器。这样做的好处是我们可以对CopyOnWrite容器进行并发的读，而不需要加锁，因为当前容器不会添加任何元素。所以CopyOnWrite容器也是一种读写分离的思想，读和写不同的容器。

//add的时候，使用重入锁，保证只有一个线程在修改
public boolean add(E e) {
        final ReentrantLock lock = this.lock;
        lock.lock();
        try {
            Object[] elements = getArray();
            int len = elements.length;
            Object[] newElements = Arrays.copyOf(elements, len + 1);
            newElements[len] = e;
            setArray(newElements);
            return true;
        } finally {
            lock.unlock();
        }
}
//get并未使用任何锁，所以在一个线程add的同时，其他线程均可以进行get操作
public E get(int index) {
        return get(getArray(), index);
}

• 应用场景
  读多写少的并发场景。

• 缺点
  内存占用问题：因为CopyOnWrite的写时复制机制，所以在进行写操作的时候，内存里会同时驻扎两个对象的内存，旧的对象和新写入的对象。如果这些对象占用的内存比较大，很有可能造成频繁的Yong GC和Full GC。针对内存占用问题，可以通过压缩容器中的元素的方法来减少大对象的内存消耗，比如，如果元素全是10进制的数字，可以考虑把它压缩成36进制或64进制。或者不使用CopyOnWrite容器，而使用其他的并发容器，如ConcurrentHashMap。
  数据一致性问题：CopyOnWrite容器只能保证数据的最终一致性，不能保证数据的实时一致性。所以如果你希望写入的的数据，马上能读到，请不要使用CopyOnWrite容器。

• CopyOnWriteArrayList

• java.util.ArrayList的线程安全版本：所有的修改操作都是通过对底层数组的最新copy来实现。

• 允许null值：All elements are permitted, including null

• 当遍历操作远多于改变操作的时候，这是一种虽然很消耗内存但非常高效的一种选择，尤其是在你不想或者不能对遍历操作进行同步处理，但是需要排除其他线程的干扰的时候。
  迭代方法使用的是迭代开始时的数组的引用，这个数组在该迭代器迭代的整个过程中都不会改变，所以该接口和迭代器不会抛出ConcurrentModificationException。该迭代器不会反应出该迭代器创建之后对应list的所有增加、移除、修改等操作。迭代器自身的元素改变方法，如remove、set和add都是不支持的。

• CopyOnWriteArraySet
  所有操作在内部使用CopyOnWriteArrayList的java.util.Set。它的特性有：

• 适用场景：set元素数量少，只读操作远多于修改操作，在遍历的时候需要排除其他线程的干扰。

• 线程安全

• add、set、remove等修改操作非常昂贵，因为当执行这些操作的时候需要对整个底层数组进行copy。

• 不支持迭代器自身的remove操作。

• 通过iterators进行迭代时的速度快，并且不受其他线程的干扰。但迭代器只会对迭代器创建时那个时刻的集合中的元素进行迭代。

这些特性也几乎是CopyOnWrite容器的特性。

2.2.2 ConcurrentMap的实现类

• ConcurrentHashMap

• 为什么需要ConcurrentHashMap
  引用自java并发编程之ConcurrentHashMap
  HashMap线程不安全（主要发生在put等对HashEntry有直接写操作的地方），Hashtable线程安全，但效率低下（Hashtable是用synchronized关键字来保证线程安全的，由于synchronized的机制是在同一时刻只能有一个线程操作，其他的线程阻塞或者轮询等待，在线程竞争激烈的情况下，这种方式的效率会非常的低下）。
  ConcurrentHashMap是高效的线程安全的：分段锁技术。Hashtable低效主要是因为所有访问Hashtable的线程都争夺一把锁。如果容器有很多把锁，每一把锁控制容器中的一部分数据，那么当多个线程访问容器里的不同部分的数据时，线程之前就不会存在锁的竞争，这样就可以有效的提高并发的访问效率。这也正是ConcurrentHashMap使用的分段锁技术。将ConcurrentHashMap容器的数据分段存储，每一段数据分配一个Segment（锁），当线程占用其中一个Segment时，其他线程可正常访问其他段数据。

• 完全支持检索和修改时的并发。该类与java.util.Hashtable具有同样的功能定义，并且对Hashtable的每个方法都有相应的实现。但是，尽管所有的操作都是线程安全的，检索操作并不意味着加锁，并不对整个表进行加锁（这样会阻止所有访问）。就线程安全而言，该类与Hashtable是相通的，只是在实现线程同步的细节上不同。

• 检索操作（包括get）通常并不会阻塞，所以可能会和更新操作（包括put和remove）同时进行。检索操作反映了最近一次更新后的结果。对于特定key的检索操作，总能得到更新后的value值。但是对于putAll和clear操作，并发的检索可能只会反应一部分更新结果。同样的，Iterators, Spliterators 和 Enumerations返回的是对应的iterator/enumeration刚创建时的结果。它们不会抛出ConcurrentModificationException异常。但是，iterators被设计成在某一时刻只能被一个线程使用。要记住：size、isEmpty、containsValue通常只在其他线程没有对该map进行修改时使用，否则，它们的结果只能反应出修改操作过程中的某个精确值，并不能反应最终结果。

• 当有太多冲突的时候，哈希表会动态扩容。当这是一个相当慢的操作，所以，尽可能地在构造函数中提供一个size来作为初始容量值。要注意的是，对于任何哈希表而言，使用太多哈希值相同的key都会明显对性能造成不利影响。为了消除这种影响，如果key实现了Comparable接口，该类会使用keys之间的比较顺序来help break ties。

• key和value都不允许为null

• ConcurrentSkipListMap
  引用自JDK并发工具类源码学习系列——ConcurrentSkipListMap

• 为什么需要ConcurrentSkipListMap
  该类使用范围不是很广，它是针对某一特殊需求而设计的——支持排序，同时支持搜索目标返回最接近匹配项的导航方法。
  与TreeMap相比，两者都是有序的哈希表，但是TreeMap是非线程安全的（可以使用Collections.synchronizedSortedMap将TreeMap进行包装，但是性能不高，毕竟多个线程一起读都需要加锁是不太合理的，至少做到读写分离呀）ConcurrentSkipListMap。这是一个支持高并发度的有序哈希表，并且是无锁的，可在多线程环境下替代TreeMap。

• 实现原理
  ConcurrentSkipListMap使用SkipList(跳表)实现排序，而TreeMap使用红黑树

2.2.3 阻塞队列

• 概念
  阻塞队列是一个支持阻塞插入和阻塞移除的队列：当队列满时，队列会阻塞插入元素的线程，直到队列不满；当队列为空时，队列会阻塞获取元素的线程，直到队列不空。
  阻塞队列常用于生产者和消费者模式，生产者向队列中添加元素，消费者则从队列中取出元素。
• 4种处理方式

方法/方式处理	抛出异常	返回特殊值	一直阻塞	超时退出
插入	add(e)	offer(e)	put(e)	offer(e, time, unit)
移除	remove()	poll()	take()	poll(time, unit)
检查	element()	peek()	不可用	不可用

①如果队列已满：
使用add(e)添加元素，则抛出IllegalStateException("Queue full")异常；
使用offer(e)添加元素，则返回false；
使用put(e)添加元素，则当前线程会一直阻塞，直到队列中出现空位或响应中断退出；
使用offer(e, time, unit)添加元素，则当前线程会阻塞一定时间，超时后如果还是满，则返回false，如果在超时前放入成功，则返回true

②如果队列为空：
使用remove()移除元素，则抛出NoSuchElementException异常；
使用poll()移除元素，则返回null；
使用take()移除元素，则当前线程会一直阻塞，直到队列中有元素插入或响应中断退出；
使用poll(time, unit)移除元素，则当前线程会阻塞一定时间，超时后如果还是为空，则返回null，如果在超时前有元素插入，则返回插入的这个元素

• 7种阻塞队列

  • ArrayBlockingQueue
    使用数组实现的有界阻塞队列，按照FIFO的原则对元素排序；内部使用重入锁可实现公平访问。内部使用一个重入锁来控制并发修改操作，即同一时刻，只能进行放或取中的一个操作。初始化时，必须指定容量大小。
  • LinkedBlockingQueue
    使用链表实现的有界阻塞队列，按照FIFO的原则对元素排序；默认和最大长度均为Integer.MAX_VALUE，所以在使用的时候，要注意指定最大容量，否则可能会导致元素数量过多，内存溢出。内部使用两个重入锁来控制并发操作，即同一时刻，允许同时进行放和取。
  • PriorityBlockingQueue
    支持优先级的无界阻塞队列，默认情况下元素按照自然顺序升序排列，可以自定义类实现compareTo()方法来指定元素的排序规则，或在初始化PriorityBlockingQueue时指定构造参数Comparator来对元素进行排序，但不能保证同优先级元素的顺序；
  • DelayQueue
    支持延时获取元素的无界阻塞队列，队列中的元素必须实现Delayed接口，在创建元素时可以指定多久才能从队列中获取当前元素，只有在延迟期满后，才能从队列中获取元素。
    DelayQueue可以应用在缓存系统的设计（用DelayQueue保存缓存元素的有效期，使用一个线程循环查询DelayQueue，一旦能从DelayQueue中获取元素，表示缓存有效期到了）、定时任务调度等场景（ScheduledThreadPoolExecutor中的ScheduledFutureTask类就是实现的Delayed接口）
  • SyncronousQueue
    不存储元素的阻塞队列，每一个put操作必须等待一个take操作，否则不能继续添加元素，支持公平访问队列，非常适合传递性场景，即把生产者线程处理的数据直接传递给消费者线程，队列本身不存储任何元素。SyncronousQueue的吞吐量高于ArrayBlockingQueue和LinkedBlockingDeque。
  • LinkedTransferQueue
    使用链表实现的无界阻塞TransferQueue，当有消费者正在等待接受元素时，队列可以通过transfer()方法把生产者传入的元素立即传给消费者。
  • LinkedBlockingDeque
    使用链表实现的双向阻塞队列，可以在队列的两端进行插入和移除元素。

2.2.4 其他实现类（待补充）

• ConcurrentLinkedQueue
  ConcurrentLinkedQueue适合在对性能要求相对较高，同时对队列的读写存在多个线程同时进行的场景，即如果对队列加锁的成本较高则适合使用无锁的ConcurrentLinkedQueue来替代。LinkedBlockingQueue 多用于任务队列，ConcurrentLinkedQueue 多用于消息队列
• ConcurrentLinkedDeque
• ConcurrentSkipListSet
  可以理解为TreeSet的线程安全并且并发性能好的版本

---

参考

未一一列出

20171210更新：增加阻塞队列的介绍
20171211更新：增加CopyOnWrite容器的add和get方法源码及说明
20171220更新：更新阻塞队列中ArrayBlockingQueue和LinkedBlockingQueue的介绍