引子
1、不安全：大家都知道HashMap不是线程安全的，在多线程环境下，对HashMap进行put操作会导致死循环。是因为多线程会导致Entry链表形成环形数据结构，这样Entry的next节点将永远不为空，就会产生死循环获取Entry。具体内容见HashMap随笔。
2、不高效：Collections.synchronizedMap(hashMap)和HashTable的线程安全原理都是对方法进行同步，所有操作竞争同一把锁，性能比较低。
如何构造一个线程安全且高效的HashMap？ConcurrentHashMap登场。
锁分段技术
ConcurrentHashMap将数据分为很多段（Segment），Segment继承了ReentrantLock，每个段都是一把锁。每个Segment都包含一个HashEntry数组，HashEntry数组存放键值对数据。当一个线程要访问Entry数组时，需要获取所在Segment锁，保证在同一个Segment的操作是线程安全的，但其他Segment的数据的访问不受影响，可以实现并发的访问不同的Segment。同一个段中才存在竞争关系，不同的段之间没有竞争关系。
ConcurrentHashMap源码分析
源码分析基于jdk1.7，不同版本实现有所不同。
类图
初始化
segmentShift和segmentMask的作用是定位Segment索引。以默认值为例，concurrencyLevel为16，需要移位4次（sshift为4），segmentShift就等于28，segmentMask等于15。
concurrencyLevel是指并发级别，即Segment数组的大小。concurrencyLevel值得设定应该根据并发线程数决定。如果并发级别设置的太小，同一个Segment的元素数量过多，会引起锁竞争的加重；如果太大，原本属于同一个Segment的元素会被分配到不同的Segment，会引起Cpu缓存命中率下降，进而导致程序性能下降。
//initialCapacity:初始容量，默认16。
//loadFactor:负载因子，默认0.75。当元素个数大于loadFactor*最大容量时需要扩容（rehash）
//concurrencyLevel:并发级别，默认16。确定Segment的个数，Segment的个数为大于等于concurrencyLevel的第一个2^n。
public ConcurrentHashMap(int initialCapacity,
float loadFactor, int concurrencyLevel) {
//判断参数是否合法
if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0)
throw new IllegalArgumentException();
//Segment最大个数MAX_SEGMENTS = 1 << 16,即65536;
if (concurrencyLevel > MAX_SEGMENTS)
concurrencyLevel = MAX_SEGMENTS; // Find power-of-two sizes best matching arguments
int sshift = 0;
int ssize = 1;
//使用循环找到大于等于concurrencyLevel的第一个2^n。ssize就表示Segment的个数。
while (ssize < concurrencyLevel) {
++sshift; //记录移位的次数，
ssize <<= 1;//左移1位
}
this.segmentShift = 32 - sshift; //用于定位hash运算的位数，之所以用32是因为ConcurrentHashMap里的hash()方法输出的最大数是32位的
this.segmentMask = ssize - 1; //hash运算的掩码，ssize为2^n，所以segmentMask每一位都为1。目的是之后可以通过key的hash值与这个值做&运算确定Segment的索引。
//最大容量MAXIMUM_CAPACITY = 1 << 30;
if (initialCapacity > MAXIMUM_CAPACITY)
initialCapacity = MAXIMUM_CAPACITY;
//计算每个Segment所需的大小，向上取整
int c = initialCapacity / ssize;
if (c * ssize < initialCapacity)
++c;
int cap = MIN_SEGMENT_TABLE_CAPACITY;//每个Segment最小容量MIN_SEGMENT_TABLE_CAPACITY = 2;
//cap表示每个Segment的容量，也是大于等于c的2^n。
while (cap < c)
cap <<= 1;
//创建一个Segment实例，作为Segment数组ss的第一个元素
// create segments and segments[0]
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;
}
插入元素（put）
可以分为三步：
1、定位Segment：通过Hash值与segmentShift、segmentMask的计算定位到对应的Segment；
2、锁获取：获取对应Segment的锁，如果获取锁失败，需要自旋重新获取锁；如果自旋超过最大重试次数，则阻塞。
3、插入元素：如果key已经存在，直接更新；如果key不存在，先判断是否需要扩容，若需要则执行rehash()后插入原因，否则直接存入元素。
为了高效，ConcurrentHashMap不会对整个容器进行扩容，而只对某个segment进行扩容。
Segment的扩容判断比HashMap更恰当，因为HashMap是在插入元素后判断元素是否已经到达容量的，如果到达了就进行扩容，但是很有可能扩容之后没有新元素插入，这时HashMap就进行了一次无效的扩容。
与HashMap不同ConcurrentHashMap并不允许key或者value为null。
/**ConcurrentHashMap中方法**/
public V put(K key, V value) {
Segment<K,V> s;
if (value == null)
throw new NullPointerException();
int hash = hash(key); //计算hash值，hash值是一个32位的整数
//计算Segment索引
//在默认情况下，concurrencyLevel为16，segmentShift为28，segmentMask为15。
//先右移28位，hash值变为0000 0000 0000 0000 0000 0000 0000 xxxx，
//与segmentMask做&运算，就是取最后四位的值。这个值就是Segment的索引
int j = (hash >>> segmentShift) & segmentMask;
//通过UNSAFE的方式获取索引j对应的Segment对象。
if ((s = (Segment<K,V>)UNSAFE.getObject // nonvolatile; recheck
(segments, (j << SSHIFT) + SBASE)) == null) // in ensureSegment
//Segment采用延迟初始化机制，如果sement为null，则调用ensureSegment创建Segment
s = ensureSegment(j);
//向Segment中put元素
return s.put(key, hash, value, false);
} /**ConcurrentHashMap$Segment中方法**/
//向Segment中put元素
final V put(K key, int hash, V value, boolean onlyIfAbsent) {
//获取锁。如果获取锁成功，插入元素，和普通的hashMap差不多。
//如果获取锁失败，执行scanAndLockForPut进行重试。重试设计见scanAndLockForPut方法源码。
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数组索引
HashEntry<K,V> first = entryAt(tab, index);
for (HashEntry<K,V> e = first;;) {
if (e != null) { //该索引处已经有元素
K k; //如果key相同，替换value。
if ((k = e.key) == key ||
(e.hash == hash && key.equals(k))) {
oldValue = e.value;
//onlyIfAbsent=true参数表示如果key存在，则不更新value值，只有在key不存在的情况下，才更新。
//在putIfAbsent方法中onlyIfAbsent=true
//在put方法中onlyIfAbsent=false
if (!onlyIfAbsent) {Scans
e.value = value;
++modCount;//修改次数
}
break;
}
e = e.next;//继续找下一个元素
}
else {
if (node != null)
node.setNext(first);
else
node = new HashEntry<K,V>(hash, key, value, first);
int c = count + 1; //count为ConcurrentHashMap$Segment中的域
if (c > threshold && tab.length < MAXIMUM_CAPACITY)
//如果元素数量超过阈值且表长度小于MAXIMUM_CAPACITY，扩容
rehash(node);
else
setEntryAt(tab, index, node);//将node节点更新到table中
++modCount;
count = c;
oldValue = null;
break;
}
}
} finally {
unlock();
}
return oldValue;
} /**ConcurrentHashMap$Segment中方法**/
//自旋获取锁
private HashEntry<K,V> scanAndLockForPut(K key, int hash, V value) {
//entryForHash根据hash值找到当前segment中对应的HashEntry数组索引。
HashEntry<K,V> first = entryForHash(this, hash);
HashEntry<K,V> e = first;
HashEntry<K,V> node = null;
int retries = -1; // negative while locating node
//自旋获取锁。若获取到锁，则跳出循环；否则一直循环直到获取到锁或retries大于MAX_SCAN_RETRIES。
while (!tryLock()) {
HashEntry<K,V> f; // to recheck first below
//当retries = -1时（即第一次循环或更新操作导致的first节点发生变化），会遍历该Segment的HashEntry数组中hash对应的链表，如果key对应的HashEntry不存在，则创建该节点。
//此处遍历链表的原因：希望遍历的链表被CPU cache所缓存，为后续实际put过程中的链表遍历操作提升性能。怎么理解呢？放在put时再去遍历不行吗？因为此时当前线程没有获取到Segment锁，所以不能进行put操作，但可以为put操作做一些准备工作（有可能加载到缓存），使put的操作更快，从而减少锁竞争。这种思想在remove()方法中也有体现。
if (retries < 0) {
if (e == null) {
//如果key不存在创建node，然后进入下一个循环
if (node == null) // speculatively create node
node = new HashEntry<K,V>(hash, key, value, null);
retries = 0;
}
else if (key.equals(e.key))
//如果key存在直接进入下一个循环
retries = 0;
else
e = e.next; //链表的下一个节点
}
else if (++retries > MAX_SCAN_RETRIES) {
//每次循环，retries加1，判断是否大于最大重试次数MAX_SCAN_RETRIES.
//static final int MAX_SCAN_RETRIES = Runtime.getRuntime().availableProcessors() > 1 ? 64 : 1;
//为了防止自旋锁大量消耗CPU的缺点。如果超过MAX_SCAN_RETRIES，使用lock方法获取锁。如果获取不到锁则当前线程阻塞并跳出循环。
//ReentrantLock的lock()和tryLock()方法的区别。
lock();
break;
}
else if ((retries & 1) == 0 &&
(f = entryForHash(this, hash)) != first) {
//每隔一次循环，检查所在数组索引的链表头结点有没有变化（其他线程有更新Map的操作，如put，rehash或者remove操作）。
//如果改变，retries更新为-1，重新遍历
e = first = f; // re-traverse if entry changed
retries = -1;
}
}
return node;
} /**ConcurrentHashMap$Segment中方法**/
//rehash
private void rehash(HashEntry<K,V> node) {
HashEntry<K,V>[] oldTable = table;
int oldCapacity = oldTable.length;
int newCapacity = oldCapacity << 1; //新容量为旧容量的2倍
threshold = (int)(newCapacity * loadFactor); //新阈值
HashEntry<K,V>[] newTable =
(HashEntry<K,V>[]) new HashEntry[newCapacity]; //新表
int sizeMask = newCapacity - 1; //新掩码
//对旧表做遍历
for (int i = 0; i < oldCapacity ; i++) {
HashEntry<K,V> e = oldTable[i];
if (e != null) {
HashEntry<K,V> next = e.next;
int idx = e.hash & sizeMask;
if (next == null) // Single node on list 链表中只存在一个节点
newTable[idx] = e;
else { // Reuse consecutive sequence at same slot
//链表中存在多个节点.
/*
相对于HashMap的resize，ConcurrentHashMap的rehash原理类似，但是Doug Lea为rehash做了一定的优化，避免让所有的节点都进行复制操作：由于扩容是基于2的幂指来操作，假设扩容前某HashEntry对应到Segment中数组的index为i，数组的容量为capacity，那么扩容后该HashEntry对应到新数组中的index只可能为i或者i+capacity，因此大多数HashEntry节点在扩容前后index可以保持不变。基于此，rehash方法中会定位第一个后续所有节点在扩容后index都保持不变的节点，然后将这个节点之前的所有节点重排即可
*/
HashEntry<K,V> lastRun = e;
int lastIdx = idx;
//找到第一个在扩容后index都保持不变的节点lastRun
for (HashEntry<K,V> last = next;
last != null;
last = last.next) {
int k = last.hash & sizeMask;
if (k != lastIdx) {
lastIdx = k;
lastRun = last;
}
}
newTable[lastIdx] = lastRun;
// Clone remaining nodes
//将这个节点之前的所有节点重排
for (HashEntry<K,V> p = e; p != lastRun; p = p.next) {
V v = p.value;
int h = p.hash;
int k = h & sizeMask;
HashEntry<K,V> n = newTable[k];
newTable[k] = new HashEntry<K,V>(h, p.key, v, n);
}
}
}
}
int nodeIndex = node.hash & sizeMask; // add the new node
node.setNext(newTable[nodeIndex]);
newTable[nodeIndex] = node;
table = newTable;
}
Segment延迟初始化机制
Segment采用延迟初始化机制，如果sement为null，则调用ensureSegment确保创建Segment。
ensureSegment方法可能被多个线程调用，ensureSegment()是怎么保证线程安全的呢？
通过源代码可看出ensureSegment方法并未使用锁来控制竞争，而是使用了Unsafe对象的getObjectVolatile()提供的原子读语义结合CAS来确保Segment创建的原子性。
ensureSegment()源代码：
@SuppressWarnings("unchecked")
private Segment<K,V> ensureSegment(int k) {
final Segment<K,V>[] ss = this.segments;
long u = (k << SSHIFT) + SBASE; // raw offset
Segment<K,V> seg;
if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) {
//使用第一个segment作为模板来创建segment，第一个segment在Map初始化时已经被创建
Segment<K,V> proto = ss[0]; // use segment 0 as prototype
int cap = proto.table.length;
float lf = proto.loadFactor;
int threshold = (int)(cap * lf);
HashEntry<K,V>[] tab = (HashEntry<K,V>[])new HashEntry[cap];
if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
== null) { // recheck
Segment<K,V> s = new Segment<K,V>(lf, threshold, tab); //根据第一个segment的参数创建新的Segment
//自旋CAS。如果seg!=null,说明该segment已经被其他线程创建，则方法结束；如果seg==null，说明该segment还没有被创建，则当前线程采用CAS更新Segment数组，如果CAS成功，则结束，否则说明其他线程对Segment数组有过更新，继续下一个循环指定该segment创建成功。
while ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
== null) {
if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s))
break;
}
}
}
return seg;
}
scanAndLockForPut方法
自旋获取锁中，当第一次循环或更新操作导致的first节点发生变化时，会遍历该Segment的HashEntry数组中hash对应的链表，如果key对应的HashEntry不存在，则创建该节点。
此处遍历链表的原因：希望遍历的链表被CPU cache所缓存，为后续实际put过程中的链表遍历操作提升性能。怎么理解呢？put还是要再去遍历一次（即使链表在缓存中）？因为此时当前线程没有获取到Segment锁，所以不能进行put操作，但可以为put操作做一些准备工作（有可能加载到缓存，在缓存中执行遍历更快），使put的操作更快，从而减少锁竞争。这种思想在remove()方法中也有体现。
获取元素（get）
get操作不需要加锁，当拿到的值为空时才会加锁重读。读操作不用加锁的原因是它的get方法里将要使用的共享变量都定义成volatile类型，如volatile V value。定义成volatile的变量，能够在线程之间保持可见性，能够被多线程同时读，并且保证不会读到过期的值，但是只能被单线程写（有一种情况可以被多线程写，就是写入的值不依赖于原值）。get方法使用UNSAFE提供的原子读语义来获的Segmnet和对应的链表。
containsKey方法和get相似，都不用加锁。
public V get(Object key) {
Segment<K,V> s; // manually integrate access methods to reduce overhead
HashEntry<K,V>[] tab;
int h = hash(key);
long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
//通过Hash值找到相应的Segment
if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null &&
(tab = s.table) != null) {
//找到HashEntry链表的索引，遍历链表找到对应的key
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;
}
统计大小（size）
统计Map的大小需要统计所有Segment的大小然后求和。
问题：累加的过程中Segment的大小可能会发生变化，导致统计的结果不准确。
解决方案：1）简单的方法就是对所有的Segment加锁，但方法低效。
2）考虑到累加的过程中Segment的大小变化的可能性很小，作者给出了更高效的方案，首先尝试几次在不对Segment加锁的情况下统计各个Segment的大小，如果累加期间Map的大小发生了变化，再使用加锁的方式统计各个Segment的大小。判断Map的大小是否发生了变化，需要通过Segment的modCount变量实现。modCount表示对Segment的修改次数。相同的思想也用在了containsValue操作。
注意事项：使用加锁方式进行统计大小时，对每一个Segment加锁，需要强制创建所有的Segment，这么做的目的是防止其他线程创建Segment并进行更新操作。所以应尽量避免在多线程环境下使用size和containsValue方法。
public int size() {
// Try a few times to get accurate count. On failure due to
// continuous async changes in table, resort to locking.
final Segment<K,V>[] segments = this.segments;
int size;
boolean overflow; // true if size overflows 32 bits
long sum; // sum of modCounts
long last = 0L; // previous sum
int retries = -1; // first iteration isn't retry
try {
for (;;) {
//static final int RETRIES_BEFORE_LOCK = 2;
//判断是否到达无锁统计map大小的最大次数，若达到最大次数需要锁所有Segment
if (retries++ == RETRIES_BEFORE_LOCK) {
//对每一个Segment加锁，此时需要强制创建所有的Segment，这么做的目的是防止其他线程创建Segment并进行更新操作。
//所以应避免在多线程环境下使用size和containsValue方法。
for (int j = 0; j < segments.length; ++j)
ensureSegment(j).lock(); // force creation
}
sum = 0L;
size = 0;
overflow = false;
for (int j = 0; j < segments.length; ++j) {
Segment<K,V> seg = segmentAt(segments, j);
if (seg != null) {
sum += seg.modCount;
int c = seg.count;
if (c < 0 || (size += c) < 0)
overflow = true;
}
}
//判断前后两次统计的modCount之和是否相等，若相等则说明没有被修改郭
//由于last初始值为0，如果该Map从创建到现在都没有被修改过，即所有Segment的modCount都为0，则只执行一次循环；否则至少执行两次循环，比较两次统计的sum有没有发生变化。又因为retries初始值-1，所以可以说重试无锁统计大小的次数为3次。
if (sum == last)
break;
last = sum;
}
} finally {
//重试次数大于最大次数，需要释放锁
if (retries > RETRIES_BEFORE_LOCK) {
for (int j = 0; j < segments.length; ++j)
segmentAt(segments, j).unlock();
}
}
return overflow ? Integer.MAX_VALUE : size;
}
Java8的ConcurrentHashMap
相对于Java7中的实现，主要有以下两点改进：
1）取消segment分段，直接使用数组transient volatile Node<K,V>[] table存储数据，将table数组元素作为锁，实现对数组中每一个桶进行加锁，进一步减少并发冲突的概率。
2）类似于Java8中的HashMap，将数组+链表的结构变更为数组+链表+红黑树的结构。当链表的长度大于8时，将链表转换为红黑树，原因见HashMap。
通过 Node + CAS + Synchronized 来保证线程安全。
Fields
transient volatile Node<K,V>[] table;//存放元素的数组，懒加载，大小是2的n次方
private transient volatile Node<K,V>[] nextTable;//扩容时用到
//基本计数器，通过CAS更新
private transient volatile long baseCount;
/*控制标识符，用来控制table的初始化和扩容的操作，不同的值有不同的含义
*当为负数时：-1代表正在初始化，-N代表有N-1个线程正在进行扩容
*当为0时：代表当时的table还没有被初始化
*当为正数时：表示初始化或者下一次进行扩容的大小
*/
private transient volatile int sizeCtl; /**
* The next table index (plus one) to split while resizing.
*/
private transient volatile int transferIndex; /**
* Spinlock (locked via CAS) used when resizing and/or creating CounterCells.
*/
private transient volatile int cellsBusy; /**
* Table of counter cells. When non-null, size is a power of 2.
*/
private transient volatile CounterCell[] counterCells; // views
private transient KeySetView<K,V> keySet;
private transient ValuesView<K,V> values;
private transient EntrySetView<K,V> entrySet;
Node
hash值和key都是final的，不可更改；val和next都是volatile的保证可见性和禁止指令重排序。
static class Node<K,V> implements Map.Entry<K,V> {
//hash值和key都是final的，不可更改；val和next都是volatile的保证可见性和禁止指令重排序。
final int hash;
final K key;
volatile V val;
volatile Node<K,V> next;
Node(int hash, K key, V val, Node<K,V> next) {
this.hash = hash;
this.key = key;
this.val = val;
this.next = next;
}
public final K getKey() { return key; }
public final V getValue() { return val; }
public final int hashCode() { return key.hashCode() ^ val.hashCode(); }
public final String toString(){ return key + "=" + val; }
//不允许更改值
public final V setValue(V value) {
throw new UnsupportedOperationException();
}
public final boolean equals(Object o) {
Object k, v, u; Map.Entry<?,?> e;
return ((o instanceof Map.Entry) &&
(k = (e = (Map.Entry<?,?>)o).getKey()) != null &&
(v = e.getValue()) != null &&
(k == key || k.equals(key)) &&
(v == (u = val) || v.equals(u)));
}
//用于map中的get（）方法，子类重写
Node<K,V> find(int h, Object k) {
Node<K,V> e = this;
if (k != null) {
do {
K ek;
if (e.hash == h &&
((ek = e.key) == k || (ek != null && k.equals(ek))))
return e;
} while ((e = e.next) != null);
}
return null;
}
}
put
添加元素的大致过程如下：
1）如果table没有初始化，先通过initTable()方法进行初始化；
2）计算hash值，找到对应的桶，如果该桶的首节点f为null（即不存在hash冲突），使用CAS直接将新Node放入该桶；
3）如果首节点f的hash值为MOVED，说明正在扩容，先进行扩容；
4）如果存在hash冲突，则通过加锁（获取首节点f的监视器锁）来保证线程安全，分两种情况：链表和红黑树；如果是链表，则遍历链表，存在相同的key就进行覆盖，否则插入到链表的尾部；如果是红黑树，则向红黑树中插入新节点；
5）判断是否需要将链表转化为红黑树，如果需要，调用treeifyBin方法；
6）如果添加成功，就调用addCount方法统计size，并检查是否需要扩容。
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();
int hash = spread(key.hashCode());//计算hash值
//用于记录相应链表的长度
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();
//找到该hash值对应的下标i，得到第一个节点f
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
//如果第一个节点f为null，使用CAS直接将新Node放入该桶
//如果CAS成功，跳出循环结束；如果失败，进入下一个循环
if (casTabAt(tab, i, null,
new Node<K,V>(hash, key, value, null)))
break; // no lock when adding to empty bin
}
//如果f的哈希值为MOVED，则进行数据迁移（扩容）
else if ((fh = f.hash) == MOVED)
tab = helpTransfer(tab, f);
else {
//这种情况下，说明f是第一个节点且不为null
V oldVal = null;
//获取该桶第一个节点f的监视器锁
synchronized (f) {
if (tabAt(tab, i) == f) {
//第一个节点f的hash值大于0，说明是链表
if (fh >= 0) {
binCount = 1;
//遍历链表，
for (Node<K,V> e = f;; ++binCount) {
K ek;
//如果找到同样的key，判断onlyIfAbsent然后进行覆盖，跳出循环
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;
//如果遍历到链表尾部没有找到相同的key，则将新Node插入到链表的尾部
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;
}
}
}
}
//binCount != 0说明上面做了链表操作
if (binCount != 0) {
//判断是否将链表转化为红黑树，TREEIFY_THRESHOLD为8
if (binCount >= TREEIFY_THRESHOLD)
//可能转化为红黑树
//如果当前数组的长度小于64，那么会选择进行数组扩容，而不是转换为红黑树
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
addCount(1L, binCount);
return null;
}
initTable
初始化table
private final Node<K,V>[] initTable() {
Node<K,V>[] tab; int sc;
while ((tab = table) == null || tab.length == 0) {
//如果sizeCtl小于0，说明其他线程已经初始化了
if ((sc = sizeCtl) < 0)
//yield()使线程由运行状态变为就绪状态，把CPU让出来，让自己或者其它的线程运行。
Thread.yield(); // lost initialization race; just spin
//通过CAS操作将sizeCtl设置为-1，返回true代表抢到锁
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
try {
if ((tab = table) == null || tab.length == 0) {
//默认容量DEFAULT_CAPACITY为16
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
//初始化指定容量的数组，并赋给table，table为volatile的
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
table = tab = nt;
//>>>为无符号右移运算，无符号右移2位，相当于除以2
//即相当于sc=0.75n
sc = n - (n >>> 2);
}
} finally {
//将sc赋值给sizeCtl
sizeCtl = sc;
}
break;
}
}
return tab;
}
treeifyBin
链表转红黑树
private final void treeifyBin(Node<K,V>[] tab, int index) {
Node<K,V> b; int n, sc;
if (tab != null) {
//如果数组长度小于MIN_TREEIFY_CAPACITY（64）的时候，进行扩容。
if ((n = tab.length) < MIN_TREEIFY_CAPACITY)
tryPresize(n << 1);
//b是该桶中的第一个节点
else if ((b = tabAt(tab, index)) != null && b.hash >= 0) {
//获取锁
synchronized (b) {
if (tabAt(tab, index) == b) {
TreeNode<K,V> hd = null, tl = null;
//遍历链表，创建一颗红黑树
for (Node<K,V> e = b; e != null; e = e.next) {
TreeNode<K,V> p =
new TreeNode<K,V>(e.hash, e.key, e.val,
null, null);
if ((p.prev = tl) == null)
hd = p;
else
tl.next = p;
tl = p;
}
//将红黑树设置到数组的相应桶中
setTabAt(tab, index, new TreeBin<K,V>(hd));
}
}
}
}
}
tryPresize
扩容，每次都是扩容为原来的2倍，size是已经翻完倍的数值。
private final void tryPresize(int size) {
int c = (size >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY :
tableSizeFor(size + (size >>> 1) + 1);//取大于1.5倍的size+1的最近的2的n次方的值
int sc;
while ((sc = sizeCtl) >= 0) {
Node<K,V>[] tab = table; int n;
//如果数组为空，先初始化数组
if (tab == null || (n = tab.length) == 0) {
n = (sc > c) ? sc : c;
if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
try {
if (table == tab) {
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
table = nt;
sc = n - (n >>> 2);
}
} finally {
sizeCtl = sc;
}
}
}
else if (c <= sc || n >= MAXIMUM_CAPACITY)
break;
else if (tab == table) {
int rs = resizeStamp(n);
if (sc < 0) {
Node<K,V>[] nt;
if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
transferIndex <= 0)
break;
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
transfer(tab, nt);
}
else if (U.compareAndSwapInt(this, SIZECTL, sc,
(rs << RESIZE_STAMP_SHIFT) + 2))
transfer(tab, null);
}
}
}
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
int n = tab.length, stride;
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
stride = MIN_TRANSFER_STRIDE; // subdivide range
if (nextTab == null) { // initiating
try {
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
nextTab = nt;
} catch (Throwable ex) { // try to cope with OOME
sizeCtl = Integer.MAX_VALUE;
return;
}
nextTable = nextTab;
transferIndex = n;
}
int nextn = nextTab.length;
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
boolean advance = true;
boolean finishing = false; // to ensure sweep before committing nextTab
for (int i = 0, bound = 0;;) {
Node<K,V> f; int fh;
while (advance) {
int nextIndex, nextBound;
if (--i >= bound || finishing)
advance = false;
else if ((nextIndex = transferIndex) <= 0) {
i = -1;
advance = false;
}
else if (U.compareAndSwapInt
(this, TRANSFERINDEX, nextIndex,
nextBound = (nextIndex > stride ?
nextIndex - stride : 0))) {
bound = nextBound;
i = nextIndex - 1;
advance = false;
}
}
if (i < 0 || i >= n || i + n >= nextn) {
int sc;
if (finishing) {
nextTable = null;
table = nextTab;
sizeCtl = (n << 1) - (n >>> 1);
return;
}
if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
return;
finishing = advance = true;
i = n; // recheck before commit
}
}
else if ((f = tabAt(tab, i)) == null)
advance = casTabAt(tab, i, null, fwd);
else if ((fh = f.hash) == MOVED)
advance = true; // already processed
else {
synchronized (f) {
if (tabAt(tab, i) == f) {
Node<K,V> ln, hn;
if (fh >= 0) {
int runBit = fh & n;
Node<K,V> lastRun = f;
for (Node<K,V> p = f.next; p != null; p = p.next) {
int b = p.hash & n;
if (b != runBit) {
runBit = b;
lastRun = p;
}
}
if (runBit == 0) {
ln = lastRun;
hn = null;
}
else {
hn = lastRun;
ln = null;
}
for (Node<K,V> p = f; p != lastRun; p = p.next) {
int ph = p.hash; K pk = p.key; V pv = p.val;
if ((ph & n) == 0)
ln = new Node<K,V>(ph, pk, pv, ln);
else
hn = new Node<K,V>(ph, pk, pv, hn);
}
setTabAt(nextTab, i, ln);
setTabAt(nextTab, i + n, hn);
setTabAt(tab, i, fwd);
advance = true;
}
else if (f instanceof TreeBin) {
TreeBin<K,V> t = (TreeBin<K,V>)f;
TreeNode<K,V> lo = null, loTail = null;
TreeNode<K,V> hi = null, hiTail = null;
int lc = 0, hc = 0;
for (Node<K,V> e = t.first; e != null; e = e.next) {
int h = e.hash;
TreeNode<K,V> p = new TreeNode<K,V>
(h, e.key, e.val, null, null);
if ((h & n) == 0) {
if ((p.prev = loTail) == null)
lo = p;
else
loTail.next = p;
loTail = p;
++lc;
}
else {
if ((p.prev = hiTail) == null)
hi = p;
else
hiTail.next = p;
hiTail = p;
++hc;
}
}
ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) :
(hc != 0) ? new TreeBin<K,V>(lo) : t;
hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) :
(lc != 0) ? new TreeBin<K,V>(hi) : t;
setTabAt(nextTab, i, ln);
setTabAt(nextTab, i + n, hn);
setTabAt(tab, i, fwd);
advance = true;
}
}
}
}
}
}
addCount
在添加完元素之后，调用addCount方法进行计数。
addCount方法主要完成两个功能：
1）对table的长度计数+1，有两种情况：一是通过修改 baseCount，二是通过使用 CounterCell。当 CounterCell 被初始化后，就优先使用他，不再使用 baseCount了；
2）检查是否需要扩容，或者是否正在扩容。如果需要扩容，就调用扩容方法，如果正在扩容，就帮助其扩容。
//从putVal传入的参数x是1，参数check为binCount，binCount>=0,默认要检查是否需要扩容
private final void addCount(long x, int check) {
CounterCell[] as; long b, s;
//如果counterCells不为null或者更新baseCount失败
if ((as = counterCells) != null ||
!U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
CounterCell a; long v; int m;
boolean uncontended = true;
//如果counterCells的大小为0，
//或者随机取其中一个元素为null，
//或者修改这个槽位的变量失败,则执行fullAddCount方法
if (as == null || (m = as.length - 1) < 0 ||
(a = as[ThreadLocalRandom.getProbe() & m]) == null ||
!(uncontended =
U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
fullAddCount(x, uncontended);
return;
}
if (check <= 1)
return;
s = sumCount();//计算map的size赋值给s
}
//判断是否需要扩容，在putVal中调用，默认都是要检查的
if (check >= 0) {
Node<K,V>[] tab, nt; int n, sc;
//如果map的size大于sizeCtl（扩容阈值），
//且table不是null，
//且table的长度小于MAXIMUM_CAPACITY,则扩容
while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
(n = tab.length) < MAXIMUM_CAPACITY) {
int rs = resizeStamp(n);
//sizeCtl小于0表示正在扩容
if (sc < 0) {
if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
transferIndex <= 0)
break;
// 如果可以帮助扩容，那么将 sc 加 1. 表示多了一个线程在帮助扩容
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
transfer(tab, nt);
}
//如果没有在扩容，将 sc 更新为负数，更新成功就进行扩容
else if (U.compareAndSwapInt(this, SIZECTL, sc,
(rs << RESIZE_STAMP_SHIFT) + 2))
transfer(tab, null);//进行扩容。
s = sumCount();
}
}
}
get
获取元素的大致过程如下：
1）计算hash值，找到数组table中对应的桶；
2）如果该桶的首节点为null，直接返回null；
3）如果该桶的首节点的key就是要找的key，直接返回其value；
4）如果该桶的首节点的hash值<0，说明正在扩容或者该位置是红黑树，通过find方法找到想要的值；
5）如果是链表，遍历链表查找相同的key。
public V get(Object key) {
Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
int h = spread(key.hashCode());//计算hash值
if ((tab = table) != null && (n = tab.length) > 0 &&
(e = tabAt(tab, (n - 1) & h)) != null) {
//如果该桶的第一个节点就是要找的key，直接返回value
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;
}
size
size = baseCount + CounterCell数组中元素的个数。因为在addCount方法中，使用CAS更新baseCount，有可能在并发情况下更新失败。即节点已经被添加到数组table中，但数量没有被统计。当更新失败时，会调用fullAddCount方法将这些失败的节点包装成一个CounterCell对象，并保存在CounterCell数组中。
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) {
for (int i = 0; i < as.length; ++i) {
if ((a = as[i]) != null)
sum += a.value;
}
}
return sum;
}
总结
Java8版本的ConcurrentHashMap相对于Java7有什么优势：
1）Java7中锁的粒度为segment，每个segment中包含多个HashEntry，而Java8中锁的粒度就是HashEntry（首节点）；
2）Java7中锁使用的是ReentrantLock，而Java8中使用的是synchronized；
为什么使用内置锁synchronized来代替重入锁ReentrantLock？
1 在低粒度的加锁方式中，synchronized的性能不比ReentrantLock差；Java8中ConcurrentHashMap的锁粒度更低了，发生冲突的概率更低，JVM对synchronized进行了大量的优化（自旋锁、偏向锁、轻量级锁等等），只要线程在可以在自旋过程中拿到锁，那么就不会升级为重量级锁，就避免了线程挂起和唤醒的上下文开销。但使用ReentrantLock不会自旋，而是直接被挂起，当然，也可以使用tryLock()，但是这样又出现了一个问题，你怎么知道tryLock的时间呢?在时间范围里还好，假如超过了呢？
所以在低粒度的加锁方式中，synchronized是最好的选择。Synchronized和ReentrantLock他们的开销差距是在释放锁时唤醒线程的数量，Synchronized是唤醒锁池里所有的线程+刚好来访问的线程，而ReentrantLock则是当前线程后进来的第一个线程+刚好来访问的线程。
2 synchronized内置锁使用起来更加简便、易懂、程序可读性好；
3 ReentrantLock需要消耗更多的内存
3）Java8中使用链表+红黑树的数据结构，代替Java7中的链表，当链表长度比较长时，红黑树的查找速度更快；
参考资料：
《java并发编程的艺术》
ConcurrentHashMap 1.8为什么要使用CAS+Synchronized取代Segment+ReentrantLock
ConcurrentHashMap原理分析（1.7与1.8）