概述
ConcurrentHashMap 是Java并发包中提供的一个线程安全且高效的HashMap实现，ConcurrentHashMap 在并发编程的场景中使用频率非常之高。
HashMap
HashMap是线程不安全的，在并发环境下，可能会造成环状链表（扩容时可能造成），导致get操作时，cpu空转达到100%，所以，在并发环境中使用HashMap是非常危险的。
HashTable
HashTable 和 HashMap 的实现原理几乎一样，差别是：1.HashTable不允许key和value为null;2.HashTable是线程安全的。但是HashTable线程安全的策略实现代价却太大了，简单粗暴，get/put所有相关操作都是synchronized的，这相当于给整个哈希表加了一把大锁，多线程访问的时候，只要有一个线程访问或操作该对象，那其他线程只能阻塞，相当于将所有的操作串行化，在竞争激烈的并发场景中性能就会非常差。
1.png
HashTable性能差主要是由于所有操作需要竞争同一把锁，而如果容器中有多把锁，每一把锁锁一段数据，这样在多线程访问不同段的数据时，就不会存在锁竞争了，这样便可以有效地提高并发效率。这就是ConcurrentHashMap所采用的“分段锁”思想。
2.png
数据结构
jdk1.7 中采用 segment+ hashEntry 的方式进行实现，结构如下：
3.png
ConcurrentHashMap 的构造函数
//初始的容量
static final int DEFAULT_INITIAL_CAPACITY = 16;
//初始的加载因子
static final float DEFAULT_LOAD_FACTOR = 0.75f;
//初始的并发等级（下面会叙述作用）
static final int DEFAULT_CONCURRENCY_LEVEL = 16;
//最大容量
static final int MAXIMUM_CAPACITY = 1 << 30;
//最小的segment数量
static final int MIN_SEGMENT_TABLE_CAPACITY = 2;
//最大的segment数量
static final int MAX_SEGMENTS = 1 << 16;
//
static final int RETRIES_BEFORE_LOCK = 2;
//通过指定的容量，加载因子和并发等级创建一个新的ConcurrentHashMap
public ConcurrentHashMap(int initialCapacity,
float loadFactor, int concurrencyLevel) {
//对容量，加载因子和并发等级做限制
if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0)
throw new IllegalArgumentException();
//限制并发等级不可以大于最大等级
if (concurrencyLevel > MAX_SEGMENTS)
concurrencyLevel = MAX_SEGMENTS;
// 下面即通过并发等级来确定Segment的大小
//sshift用来记录向左按位移动的次数
int sshift = 0;
//ssize用来记录Segment数组的大小
int ssize = 1;
//Segment的大小为大于等于concurrencyLevel的第一个2的n次方的数
while (ssize < concurrencyLevel) {
++sshift;
ssize <<= 1;
}
this.segmentShift = 32 - sshift;
//segmentMask的值等于ssize - 1（这个值很重要）
this.segmentMask = ssize - 1;
if (initialCapacity > MAXIMUM_CAPACITY)
initialCapacity = MAXIMUM_CAPACITY;
//c记录每个Segment上要放置多少个元素
int c = initialCapacity / ssize;
//假如有余数，则Segment数量加1
if (c * ssize < initialCapacity)
++c;
int cap = MIN_SEGMENT_TABLE_CAPACITY;
while (cap < c)
cap <<= 1;
//创建第一个Segment，并放入Segment[]数组中，作为第一个Segment
Segment<K,V> s0 =
new Segment<K,V>(loadFactor, (int)(cap * loadFactor),
(HashEntry<K,V>[])new HashEntry[cap]);
Segment<K,V>[] ss = (Segment<K,V>[])new Segment[ssize];
UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]
this.segments = ss;
}
从上面的代码可以看出来，Segment 数组的大小ssize是由concurrentLevel来决定的，但是却不一定等于concurrentLevel，ssize一定是大于或等于concurrentLevel的最小的2的次幂。比如：默认情况下concurrentLevel是16，则ssize为16；若concurrentLevel为14，ssize为16；若concurrentLevel为17，则ssize为32。
ConcurrentHashMap 的put操作
public V put(K key, V value) {
Segment<K,V> s;
//ConcurrentHashMap的key和value都不能为null
if (value == null)
throw new NullPointerException();
//这里对key求hash值，并确定应该放到segment数组的索引位置
int hash = hash(key);
//j为索引位置，思路和HashMap的思路一样，这里不再多说
int j = (hash >>> segmentShift) & segmentMask;
if ((s = (Segment<K,V>)UNSAFE.getObject // nonvolatile; recheck
(segments, (j << SSHIFT) + SBASE)) == null) // in ensureSegment
s = ensureSegment(j);
//这里很关键，找到了对应的Segment，则把元素放到Segment中去
return s.put(key, hash, value, false);
}
得到的hash值向右移动segmentShift位，然后在于segmentMask做&运算得到segment的索引j。例如：concurrencyLevel等于16，则sshift等于4，则segmentShift为28。hash值是一个32位的整数，将其向右移动28位就变成这个样子：0000 0000 0000 0000 0000 0000 0000 xxxx，然后再用这个值与segmentMask做&运算，也就是取最后四位的值。这个值确定Segment的索引。
如何插入到Segment中
final V put(K key, int hash, V value, boolean onlyIfAbsent) {
//这里是并发的关键，每一个Segment进行put时，都会加锁
HashEntry<K,V> node = tryLock() ? null :
scanAndLockForPut(key, hash, value);
V oldValue;
try {
//tab是当前segment所连接的HashEntry数组
HashEntry<K,V>[] tab = table;
//确定key的hash值所在HashEntry数组的索引位置
int index = (tab.length - 1) & hash;
//取得要放入的HashEntry链的链头
HashEntry<K,V> first = entryAt(tab, index);
//遍历当前HashEntry链
for (HashEntry<K,V> e = first;;) {
//如果链头不为null
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;
}
//如果没有和key相同的，一直遍历到链尾，链尾的next为null，进入到else
e = e.next;
}
else {//如果没有找到key相同的，则把当前Entry插入到链头
if (node != null)
node.setNext(first);
else
node = new HashEntry<K,V>(hash, key, value, first);
//此时数量+1
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;
}
- 首先对key进行第一次hash，通过hash值确定segment的位置
- 然后在segment内进行操作，获取锁
- 接着获取当前segment的HashEntry数组，然后对key进行第二次hash，通过hash值确定在HashEntry数组的索引位置。
- 然后对当前索引的HashEntry链进行遍历，如果有重复的key，则替换；如果没有重复的，则插入
- 关闭锁
可见，在整个put过程中，进行了2次hash操作，才最终确定key的位置。
ConcurrentHashMap Size实现
因为ConcurrentHashMap 是可以并发插入数据的，所以在准确计算元素时存在一定的难度，一般的思路是统计每个Segment对象中的元素个数，然后进行累加，但是这种方式计算出来的结果不一定是准确的，因为在计算后面几个Segment的元素是，已经计算过的Segment同时可能有数据的插入或者删除。在JDK7的实现中，采用了如下方式：
先采用不加锁的方式，连续计算元素的个数，最多计算3次：
- 如果前后两次计算结果相同，则说明计算出来的元素个数是准确的；
- 如果前后两次计算结构都不同，则给每个Segment进行加锁，再计算一次元素的个数。
网友评论