ConcurrentHashMap，正如它的名字一样，CoccurentHashMap是支持并发的HashMap（Concurrent adj. 并发的，一致的），建议先了解HashMap之后再了解它，ConcurrentHashMap也分JDK1.7版和JDK1.8版，两个版本的实现截然不同
JDK1.7
- Segment+数组+链表，采用分段锁实现并发（每个Segement一把锁）
- 并发量理论上等于Segement的数量
- 用volatile修饰实际保存的值和指针保证内存可见行，保证多线程之间单次读取或写入的原子性
成员变量
/**
* Segment 数组，存放数据时首先需要定位到具体的 Segment 中。
*/
final Segment<K,V>[] segments;
transient Set<K> keySet;
transient Set<Map.Entry<K,V>> entrySet;
Segment的定义
Segement是ReentrantLock（重入锁）的子类
static final class Segment<K,V> extends ReentrantLock implements Serializable {
private static final long serialVersionUID = 2249069246763182397L;
// 和 HashMap 中的 HashEntry 作用一样，真正存放数据的桶
transient volatile HashEntry<K,V>[] table;
transient int count;
transient int modCount;//结构化修改数量
transient int threshold;//扩容阈值
final float loadFactor;//负载因子
}
put
//根据hash找到对应的Segement
public V put(K key, V value) {
Segment<K,V> s;
if (value == null)
throw new NullPointerException();
int hash = hash(key);
int j = (hash >>> segmentShift) & segmentMask;
if ((s = (Segment<K,V>)UNSAFE.getObject // nonvolatile; recheck
(segments, (j << SSHIFT) + SBASE)) == null) // in ensureSegment
s = ensureSegment(j);
//找到该Segement后，调用该Segement的put方法
return s.put(key, hash, value, false);
}
//将元素put到Segement
final V put(K key, int hash, V value, boolean onlyIfAbsent) {
//尝试获取锁，如果tryLock()返回false说明有其他线程竞争
//则调用scanAndLockForPut自旋获取锁，并定位当前分段table中存放的位置
HashEntry<K,V> node = tryLock() ? null :
scanAndLockForPut(key, hash, value);
V oldValue;
try {
HashEntry<K,V>[] tab = table;
int index = (tab.length - 1) & hash;
HashEntry<K,V> first = entryAt(tab, index);
//遍历该table
for (HashEntry<K,V> e = first;;) {
//该条HashEntry不为空，则说明是一个链表，循环该链表
if (e != null) {
K k;
//如果在该链表中找到一样的key，则覆盖旧值并返回旧值
if ((k = e.key) == key ||
(e.hash == hash && key.equals(k))) {
oldValue = e.value;
if (!onlyIfAbsent) {
e.value = value;
++modCount;
}
break;
}
e = e.next;
//该HashEntry为空，说明是首次放入元素
} else {
if (node != null)
node.setNext(first);
else
node = new HashEntry<K,V>(hash, key, value, first);
int c = count + 1;
//判断是否需要扩容
if (c > threshold && tab.length < MAXIMUM_CAPACITY)
rehash(node);
else
setEntryAt(tab, index, node);
++modCount;
count = c;
oldValue = null;
break;
}
}
} finally {
//最后释放锁
unlock();
}
return oldValue;
}
//scanAndLockForPut的源码：
private HashEntry<K,V> scanAndLockForPut(K key, int hash, V value) {
//定位HashEntry数组位置，获取第一个节点
HashEntry<K,V> first = entryForHash(this, hash);
HashEntry<K,V> e = first;
HashEntry<K,V> node = null;
int retries = -1;//扫描次数
while (!tryLock()) {//尝试自旋获取锁（所谓自旋就是不停的循环去获取锁）
HashEntry<K,V> f;
if (retries < 0) {
if (e == null) {
if (node == null) // 构造新节点
node = new HashEntry<K,V>(hash, key, value, null);
retries = 0;
}
else if (key.equals(e.key))
retries = 0;
else
e = e.next;
}
else if (++retries > MAX_SCAN_RETRIES) { //如果尝试达到了最大尝试次数，则改为获取阻塞锁
lock();
break;
}
else if ((retries & 1) == 0 &&
(f = entryForHash(this, hash)) != first) {
e = first = f; // 首节点有变动，更新first，重新扫描
retries = -1;
}
}
return node;
}
get
get方法逻辑清晰，获取数据效率高，因为整个过程都不需要加锁，而且获取的元素是通过volatile修饰的（HashEntry的value是volatile的），保证了内存的可见性，获取的都是最新的值
public V get(Object key) {
//定义元素所在的Segement
Segment<K,V> s; // manually integrate access methods to reduce overhead
//定义Segement所在的table
HashEntry<K,V>[] tab;
//通过hash定位到具体的Segement，下面给s赋值就是找到具体的Segement
int h = hash(key);
long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null &&
(tab = s.table) != null) {
//在当前table中寻找该元素，(tab.length - 1) & h)表示通过hash确定元素在桶中的位置
for (HashEntry<K,V> e = (HashEntry<K,V>) UNSAFE.getObjectVolatile
(tab, ((long)(((tab.length - 1) & h)) << TSHIFT) + TBASE);
e != null; e = e.next) {
K k;
if ((k = e.key) == key || (e.hash == h && key.equals(k)))
return e.value;
}
}
return null;
}
JDK1.8
- 抛弃了1.7 Segement的分段锁，采用CAS和synchronized保证并发的安全性
- 数据结构采用了数组+链表的形式，和HashMap类似
- tabAt,casTabAt,setTabAt操作都是volatile的，保证线程之前的可见性
- 写入时只要hash不冲突，就不会产生并发
- 链表元素个数大于8个转化成红黑树
成员变量
//volatile确保了table对所有线程的可见性
transient volatile Node<K,V>[] table;
Node
static class Node<K,V> implements Map.Entry<K,V> {
final int hash; //该key的hash值，final修饰，不可变
final K key; //key的实际值，final修饰，不可变
volatile V val; //真正的值，volatile修饰，保证多线程之间的内存可见行
volatile Node<K,V> next; //链表，volatile修饰，保证多线程之间的内存可见行
....
}
可以看到，这两个重要的成员都被 volatile 进行了修饰，使得他们在并发的情况下数据的变动对于其他线程是可以立即
感知的
- 首先第一个 table 用 volatile 修饰，确保了在扩容时，扩容后的结果对其他线程是立即可见的
- 其次， Node 节点中的 val 以及 next 熟悉被 volatile 修饰，确保下面操作在多线程之间的数据可见性
- 对 ConcurrentHashMap 进行 put 一份数据<K, V>
- 线程A对 Map 进行 remove 操作，线程B对 Map 进行 get 操作
- 如果A 先于 B，由于 volatile 的特新，B无法 get 到被删除的数据
put
final V putVal(K key, V value, boolean onlyIfAbsent) {
if (key == null || value == null) throw new NullPointerException();
int hash = spread(key.hashCode());
int binCount = 0;
//循环操作直到成功
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh;
//判断是否需要初始化
if (tab == null || (n = tab.length) == 0)
tab = initTable();
//如果table[i]槽位上为空，则采用cas自旋锁（自旋其实就是死循环）保证插入成功
//此处可能有有疑惑：如果两个线程同时进入了casTabAt方法，其中一个修改table[i]的值，则另外一个cat操作一直得不到预期值null，岂不是死循环？？
//其实，这里存在着java内存模型happens-before的应用（通常情况下，写操作必须要happens-before读操作）
//tabAt是volatile读，casTabAt具有volatile读写相同的内存语义
//因此casTabAt所做的操作都能立即反馈会tabAt方法并且
//jdk在这里运用了for + cas更新 实现无锁的更新操作——乐观锁，相比悲观锁性能有了一定的提升
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
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;
//上面的条件都不满足，则锁定当前node写入数据
synchronized (f) {
if (tabAt(tab, i) == f) {
//hash大于0说明是链表
if (fh >= 0) {
binCount = 1;
for (Node<K,V> e = f;; ++binCount) {
K ek;
//如果找到key相同的则替换掉旧值
if (e.hash == hash &&
((ek = e.key) == key ||
(ek != null && key.equals(ek)))) {
oldVal = e.val;
if (!onlyIfAbsent)
e.val = value;
break;
}
//都没找到则在链表尾部添加
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;
}
get
public V get(Object key) {
Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
int h = spread(key.hashCode());
if ((tab = table) != null && (n = tab.length) > 0 &&
(e = tabAt(tab, (n - 1) & h)) != null) {
//table[(n - 1) & h]槽位上的hash值刚好等于传进来的hash值，则直接返回
if ((eh = e.hash) == h) {
if ((ek = e.key) == key || (ek != null && key.equals(ek)))
return e.val;
// 如果计算出来的hash值小于0，则表示该槽位挂着的是一棵红黑树，采用二叉查找树的形式搜索
} else if (eh < 0)
return (p = e.find(h, key)) != null ? p.val : null;
// 此处则确定是链表的形式，直接遍历查找比对即可
while ((e = e.next) != null) {
if (e.hash == h &&
((ek = e.key) == key || (ek != null && key.equals(ek))))
return e.val;
}
}
return null;
}
HashMap面试题
常见的关于HashMap和ConcurrentHashMap的面试题
谈谈你理解的 HashMap，讲讲其中的 get put 过程
HashMap 这样的 Key Value 在软件开发中是非常经典的结构，常用于在内存中存放数据，底层是数组加链表的形式
- 通过元素hashCode确定元素在数据中的位置，数组的长度总是2的幂次方（为了效率），当容量到达阈值(负载因子和数组长度之积)时会引发自动扩容，非常消耗性能
- 当hash冲突时通过拉链法解决冲突，即该数组纵向延伸成为一个链表，最坏的情况退化成单一链表，时间复杂度为O(N)
- 非线程安全
- put过程
- 判断数组是否初始化
- key如果是空，则放一个空值
- 计算hash得到在数组中的位置，如果该位置上有值则遍历链表是否能找到相同的key，找到则替换原来的值
- 如果桶是空的，或者没找到相同的key，则在当前位置上创建一个新的Entry放入（如果超过了阈值就扩容）
- get过程
- 如果size为0，直接返回null
- 计算key的hash找到在数组中的位置
- 遍历当前位置的链表（如果只有一个Entry那就是单个元素的链表）寻找相同的key，找到则返回
- 都没找到返回null
JDK1.8 做了什么优化？
主要是优化了查询的效率
- 链表长度超过8则转换成红黑树，增加查找的效率，时间复杂度为O(logN)
- 扩容时会均匀分配元素，而JDK1.7会原封不动的拷贝过来
- 多线程会引发链表死循环的问题已解决
是线程安全的嘛？
不是，HashMap是非线程安全的
不安全会导致哪些问题？
多个线程写HashMap时可能会导致数据不一致，特别是在JDK1.7中多线程扩容时还会引发死循环
如何解决？有没有线程安全的并发容器？
需要线程安全的HashMap可以用Collections.synchronizedMap(new HashMap)来增加同步锁，但是这样效率比较低，所以增加了线程安全的ConcurrentHashMap
ConcurrentHashMap 是如何实现的？ 1.7、1.8 实现有何不同？为什么这么做？
- JDK 1.7
- Segment+数组+链表，采用分段锁实现并发（每个Segement一把锁）
- 并发量理论上等于Segement的数量
- 用volatile修饰实际保存的值和指针保证内存可见行，保证多线程之间单次读取或写入的原子性
- JDK 1.8（提高性能）
- 抛弃了1.7 Segement的分段锁，采用CAS和synchronized保证并发的安全性
- 数据结构采用了数组 + 链表的形式，和HashMap类似
- tabAt,casTabAt,setTabAt操作都是volatile的，保证线程之前的可见性
- 写入时只要hash不冲突，就不会产生并发
- 链表元素个数大于8个转化成红黑树