[Java并发系列] 4.Java并发容器和框架
使用Java并发容器/框架可以提高并发程序的开发效率，设计出更加优秀的并发程序。Java并发容器/框架还是位于java.util.concurrent包中。常用的并发容器有如下：
- ConcurrentHashMap
- ConcurrentLinkedQueue
- 阻塞队列
1. ConcurrentHashMap 的实现原理与使用(基于1.7)
ConcurrentHashMap 是线程安全且高效的HashMap，它可以在保证线程安全的同时可以保证高效的操作。
1. HashMap在并发环境中的问题
① 在并发环境中，HashMap的put操作后可能会导致get操作产生死循环
因为HashMap是使用拉链法解决哈希冲突，即通过链表的方式存储散列到数组相同位置的值。当在并发环境中使用HashMap时，可能会导致数组中的链表形成循环链表，在之后的get操作用会使用e = e.next去判断链表中是否还有元素，而形成链表之后，e = e.next操作就会无限循环，Infinite Loop的错误就会出现。
② 并发环境下，操作HashMap的put方法可能会导致元素丢失
当在多线程下put操作时，执行addEntry(hash, key, value, i)，如果有产生哈希碰撞，导致两个线程得到同样的bucketIndex去存储，就可能会出现覆盖丢失的情况：
2. 并发环境中如何获取线程安全的哈希表结构
- 使用Collections类中的synchronizedMap(Map<K,V> m)静态方法获取线程安全的Map，然后在此map上再进行操作；
Map m = Collections.synchronizedMap(new HashMap());
-
使用HashTable，HashTable是通过在方法或者被调用的方法上添加synchronized关键字来保证线程安全。如果在线程竞争比较激烈的情况下，HashTable的效率可能是非常低下的，因为在HashTable中，线程都必须要竞争同一把锁，正因为如此，当一个线程访问HashTable的同步方法时，其他线程只能进入阻塞或者是轮询状态。
-
使用ConcurrentHashMap，ConcurrentHashMap 使用锁分段技术，既能保证Map的线程安全，也能保证Map的效率。
3. ConcurrentHashMap的结构
ConcurrentHashMap是由Segment数组结构和HashEntry数组结构组成。Segment是一种可重入锁，是由数组和链表组成，在ConcurrentHashMap中，只有一个Segment数组，一个Segment中只包含一个HashEntry数组，每个HashEntry是一个链表结构的元素；当对HashEntry数组中的数据进行修改时，必须首先获取到与它对应的Segment锁。
4. ConcurrentHashMap的常用操作源码解读(1.8)
本机jdk环境为1.8 ，就以1.8的源码来分析，之后会有1.7与1.8中的比较总结
一、put操作
public V put(K key, V value) {
return putVal(key, value, false);
}
final V putVal(K key, V value, boolean onlyIfAbsent) {
//键值对不能为空
if (key == null || value == null) throw new NullPointerException();
//两次hash，以减少冲突，使元素均匀分布
int hash = spread(key.hashCode());
int binCount = 0;
//对数组table进行无条件自循迭代
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh;
//判断数组是否为空，如果为空则再初始化(此地方与1.7中实现不同，使用懒汉模式)
if (tab == null || (n = tab.length) == 0)
tab = initTable();
//判断i位置是否有元素，若没有的话，则使用无锁化CAS的方法进行元素的插入
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
//使用CAS插入元素
if (casTabAt(tab, i, null,
new Node<K,V>(hash, key, value, null)))
break; // no lock when adding to empty bin
}
//如果需要扩容，则先进行扩容操作
else if ((fh = f.hash) == MOVED)
tab = helpTransfer(tab, f);
else {
V oldVal = null;
//上述条件都不满足，则说明存在hash冲突，则需要进行加锁操作，锁住链表或红黑树的头结点
synchronized (f) {
if (tabAt(tab, i) == f) {
if (fh >= 0) {//表示该节点位置是链表结构
binCount = 1;
//循环该节点位置的链表
for (Node<K,V> e = f;; ++binCount) {
K ek;
//先判断是否有相同的key，如果有则直接覆盖原来的value值
if (e.hash == hash &&
((ek = e.key) == key ||
(ek != null && key.equals(ek)))) {
oldVal = e.val;
if (!onlyIfAbsent)
e.val = value;
break;
}
//没有重复的key，则将元素插入到链表的尾部
Node<K,V> pred = e;
if ((e = e.next) == null) {
pred.next = new Node<K,V>(hash, key,
value, null);
break;
}
}
}
else if (f instanceof TreeBin) {//表示该节点位置是红黑树结构
Node<K,V> p;
binCount = 2;
//插入到红黑树中
if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
value)) != null) {
oldVal = p.val;
if (!onlyIfAbsent)
p.val = value;
}
}
}
}
if (binCount != 0) {
//当链表上的元素个数大于8时，就会将链表转换为红黑树
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
addCount(1L, binCount);
return null;
}
put操作流程：
- 先对key进行两次hash计算；
- 检查table是否已经初始化，如果没有初始化，那么就先进行初始化；
- 如果没有hash冲突，那就直接使用CAS插入元素；
- 如果需要扩容，就先进行扩容操作；
- 如果产生了Hash冲突，那就对链表或者红黑树头结点进行加锁，将Key/value插入到链表尾部或者按照红黑树结构进行插入；
- 在插入结束之后，需要判断table该位置的链表长度是否大于8，如果大于8时，就会将链表转换为红黑树结构。
- 最后统计size；
二、get操作
public V get(Object key) {
Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
//对key进行两次hash计算，定位到在table中的位置
int h = spread(key.hashCode());
if ((tab = table) != null && (n = tab.length) > 0 &&
(e = tabAt(tab, (n - 1) & h)) != null) {//直接读取首节点位置的元素
if ((eh = e.hash) == h) {//如果首节点元素的hash值与key的两次hash的值相等，那么就通过对象的equals()方法，来比较key与首节点的key值是否相等，如果相等，则返回；
if ((ek = e.key) == key || (ek != null && key.equals(ek)))
return e.val;
}
//hash值为负值表示正在扩容，这个时候查的是ForwardingNode的find方法来定位到nextTable进行查找，
else if (eh < 0)
return (p = e.find(h, key)) != null ? p.val : null;
//既不是首节点也不是ForwardingNode，那就往下遍历
while ((e = e.next) != null) {
if (e.hash == h &&
((ek = e.key) == key || (ek != null && key.equals(ek))))
return e.val;
}
}
return null;
}
get操作流程：
- 计算key的hash值，定位到table的索引位置，如果是首节点，那么就直接返回；
- 如果遇到正在扩容，那么就调用标志正在扩容的ForwardingNode的find方法，查找该节点，匹配就直接返回；
- 如果上述情况都不合符，那么就遍历该链表上的节点，使用key.equals(ek)去匹配key值，匹配则返回；
三、 size操作
public int size() {
long n = sumCount();
return ((n < 0L) ? 0 :
(n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :
(int)n);
}
final long sumCount() {
CounterCell[] as = counterCells; CounterCell a;
long sum = baseCount;
if (as != null) {
//将CounterCell与baseCount累加
for (int i = 0; i < as.length; ++i) {
if ((a = as[i]) != null)
sum += a.value;
}
}
return sum;
}
5. 1.7与1.8中ConcurrentHashMap的不同
- 1.8中使用HashEntry+Synchronized+CAS+红黑树控制并发，1.7中使用Segment+Reentrantlock+HashEntry来实现；1.7中锁的粒度是基于Segment的，而1.8中锁的粒度是基于HashEntry(首节点)的，所以说1.8中锁的粒度更低，
- 1.8中使用Synchronized来进行同步，所以不需要在分段，也就不需要Segment分段的数据结构，降低实现的复杂度。
- 1.8中使用红黑树来优化链表，使用长链表进行遍历时一个很耗时的过程，而基于红黑树来查找，则会大大提高查找的速率。
2. ConcurrentLinkedQueue
在并发环境中，如果要得到一个线程安全的队列，可以通过两种方式获取：一种是通过阻塞的方式获取，另一种非阻塞的方式则是通过CAS的方式实现；而ConcurrentLinkedQueue则就是使用非阻塞的方式实现的线程安全的队列。
ConcurrentLinkedQueue是一个基于链表的无界线程安全队列。当添加一个元素时，它会将元素添加到队列的尾部，当获取一个元素时，它会从队列的头部返回一个元素。
3. Java中的阻塞队列
阻塞队列：当队列满时，队列就会阻塞向队列中插入插入元素的线程，直到队列不满；当队列为空时，会阻塞获取元素的线程，直到队列非空；
阻塞队列经常用于生产者消费者的场景，生产者是向队列中插入元素的线程，消费者是从队列中取出元素的线程。所以阻塞队列就是生产者用来存放元素、消费者用来消费元素的容器。
- ArrayBlockingQueue 使用数组实现的有界阻塞队列
- LinkedBlockingQueue 使用链表实现的有界阻塞队列，默认构造函数如下：
//
public LinkedBlockingQueue() {
this(Integer.MAX_VALUE);
}
- PriorityBlockingQueue 支持优先级排序的无界阻塞队列
- DelayQueue 使用优先级队列实现的无界阻塞队列
- SynchronousQueue 不存储元素的阻塞队列，每个put操作必须要等待一个take操作，否则是不能添加元素的；
- LinkedTransferQueue 使用链表实现的无界阻塞队列
- LinkedBlockingDeque 使用链表组成的双向阻塞队列
阻塞队列的实现原理：就是使用通知模式，当生产者往满的队列中添加元素时，就会被阻塞，当消费者从队列中消费元素之后，就会通知生产者当前的队列可用；当消费者想要在一个为空的队列中消费时，就会被阻塞，当生产者在此队列中生产元素之后，就会通知消费者此队列可用。
