作者：xbmchina 链接：https://www.jianshu.com/p/5dbaa6707017
ConcurrentHashMap是Java中的一个线程安全且高效的HashMap实现。平时涉及高并发如果要用map结构，那第一时间想到的就是它。
那么我就这几个方面了解一下ConcurrentHashMap：
1）ConcurrentHashMap在JDK8里结构 2）ConcurrentHashMap的put方法、szie方法等 3）ConcurrentHashMap的扩容 4）HashMap、Hashtable、ConccurentHashMap三者的区别 5）ConcurrentHashMap在JDK7和JDK8的区别
首先来看下底层的组成结构：
其实和 1.8 HashMap 结构类似，当链表节点数超过指定阈值的话，也是会转换成红黑树的，大体结构也是一样的。
那么它到底是如何实现线程安全的？ 答案:其中抛弃了原有的Segment 分段锁，而采用了 CAS + synchronized 来保证并发安全性。至于如何实现，那我继续看一下put方法逻辑
ConcurrentHashMap最常用的方法也就是put方法和get方法，那么下面主要看代码注释，便于理解。
/** Implementation for put and putIfAbsent */ final V putVal(K key, V value, boolean onlyIfAbsent) { if (key == null || value == null) throw new NullPointerException(); //1. 计算key的hash值 int hash = spread(key.hashCode()); int binCount = 0; for (Node<K,V>[] tab = table;;) { Node<K,V> f; int n, i, fh; //2. 如果当前table还没有初始化先调用initTable方法将tab进行初始化 if (tab == null || (n = tab.length) == 0) tab = initTable(); //3. tab中索引为i的位置的元素为null，则直接使用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 } //4. 当前正在扩容 else if ((fh = f.hash) == MOVED) tab = helpTransfer(tab, f); else { V oldVal = null; synchronized (f) { if (tabAt(tab, i) == f) { //5. 当前为链表，在链表中插入新的键值对 if (fh >= 0) { binCount = 1; for (Node<K,V> e = f;; ++binCount) { K ek; 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; } } } // 6.当前为红黑树，将新的键值对插入到红黑树中 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; } } } } // 7.插入完键值对后再根据实际大小看是否需要转换成红黑树 if (binCount != 0) { if (binCount >= TREEIFY_THRESHOLD) treeifyBin(tab, i); if (oldVal != null) return oldVal; break; } } } //8.对当前容量大小进行检查，如果超过了临界值（实际大小*加载因子）就需要扩容 addCount(1L, binCount); return null; }请先看完代码注释，有个大致的了解，然后我们更加详细的学习一下：
这个put的过程很清晰，对当前的table进行无条件自循环直到put成功，可以分成以下六步流程来概述：
1、判断Node[]数组是否初始化，没有则进行初始化操作 2、通过hash定位数组的索引坐标，是否有Node节点，如果没有则使用CAS进行添加（链表的头节点），添加失败则进入下次循环。 3、检查到内部正在扩容，就帮助它一块扩容。 4、如果f！=null，则使用synchronized锁住f元素（链表/红黑树的头元素）。如果是Node（链表结构）则执行链表的添加操作；如果是TreeNode（树型结构）则执行树添加操作。 5、判断链表长度已经达到临界值8（默认值），当节点超过这个值就需要把链表转换为树结构。 6、如果添加成功就调用addCount（）方法统计size，并且检查是否需要扩容
1.spread(key,hashCode())
这方法作用重哈希，以减小Hash冲突
static final int spread(int h) { return (h ^ (h >>> 16)) & HASH_BITS; }2.initTable方法
主要作用将tab进行初始化
private final Node<K,V>[] initTable() { Node<K,V>[] tab; int sc; while ((tab = table) == null || tab.length == 0) { if ((sc = sizeCtl) < 0) // 1. 保证只有一个线程正在进行初始化操作 Thread.yield(); // lost initialization race; just spin else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) { try { if ((tab = table) == null || tab.length == 0) { // 2. 得出数组的大小 int n = (sc > 0) ? sc : DEFAULT_CAPACITY; @SuppressWarnings("unchecked") // 3. 这里才真正的初始化数组 Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n]; table = tab = nt; // 4. 计算数组中可用的大小：实际大小n*0.75（加载因子） sc = n - (n >>> 2); } } finally { sizeCtl = sc; } break; } } return tab; }为了保证能够正确初始化，在第1步中会先通过if进行判断，若当前已经有一个线程正在初始化即sizeCtl值变为-1，这个时候其他线程在If判断为true从而调用Thread.yield()让出CPU时间片。正在进行初始化的线程会调用U.compareAndSwapInt方法将sizeCtl改为-1即正在初始化的状态。另外还需要注意的事情是，在第四步中会进一步计算数组中可用的大小即为数组实际大小n乘以加载因子0.75.可以看看这里乘以0.75是怎么算的，0.75为四分之三，这里n - (n >>> 2)是不是刚好是n-(1/4)n=(3/4)n，挺有意思的吧:)。如果选择是无参的构造器的话，这里在new Node数组的时候会使用默认大小为DEFAULT_CAPACITY（16），然后乘以加载因子0.75为12，也就是说数组的可用大小为12。
3.CAS关键操作
tabAt()该方法用来获取table数组中索引为i的Node元素。 casTabAt()利用CAS操作设置table数组中索引为i的元素 setTabAt()该方法用来设置table数组中索引为i的元素
4.ConcurrentHashMap的扩容
通过判断该节点的hash值是不是等于-1（MOVED）,代码为(fh = f.hash) == MOVED，说明 Map 正在扩容。那么就帮助 Map 进行扩容。以加快速度。 如何帮助扩容呢？那要看看 helpTransfer 方法的实现。
/** * Helps transfer if a resize is in progress. */ final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) { Node<K,V>[] nextTab; int sc; // 如果 table 不是空 且 node 节点是转移类型，数据检验 // 且 node 节点的 nextTable（新 table） 不是空，同样也是数据校验 // 尝试帮助扩容 if (tab != null && (f instanceof ForwardingNode) && (nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) { // 根据 length 得到一个标识符号 int rs = resizeStamp(tab.length); // 如果 nextTab 没有被并发修改 且 tab 也没有被并发修改 // 且 sizeCtl < 0 （说明还在扩容） while (nextTab == nextTable && table == tab && (sc = sizeCtl) < 0) { // 如果 sizeCtl 无符号右移 16 不等于 rs （ sc前 16 位如果不等于标识符，则标识符变化了） // 或者 sizeCtl == rs + 1 （扩容结束了，不再有线程进行扩容）（默认第一个线程设置 sc ==rs 左移 16 位 + 2，当第一个线程结束扩容了，就会将 sc 减一。这个时候，sc 就等于 rs + 1） // 或者 sizeCtl == rs + 65535 （如果达到最大帮助线程的数量，即 65535） // 或者转移下标正在调整 （扩容结束） // 结束循环，返回 table if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 || sc == rs + MAX_RESIZERS || transferIndex <= 0) break; // 如果以上都不是, 将 sizeCtl + 1, （表示增加了一个线程帮助其扩容） if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) { // 进行转移 transfer(tab, nextTab); // 结束循环 break; } } return nextTab; } return table; }基本逻辑已在代码注释中,这里关键transfer()，那么我们继续深入了解一下
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) { int n = tab.length, stride; // 每核处理的量小于16，则强制赋值16 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]; //构建一个nextTable对象，其容量为原来容量的两倍 nextTab = nt; } catch (Throwable ex) { // try to cope with OOME sizeCtl = Integer.MAX_VALUE; return; } nextTable = nextTab; transferIndex = n; } int nextn = nextTab.length; // 连接点指针，用于标志位（fwd的hash值为-1，fwd.nextTable=nextTab） ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab); // 当advance == true时，表明该节点已经处理过了 boolean advance = true; boolean finishing = false; // to ensure sweep before committing nextTab for (int i = 0, bound = 0;;) { Node<K,V> f; int fh; // 控制 --i ,遍历原hash表中的节点 while (advance) { int nextIndex, nextBound; if (--i >= bound || finishing) advance = false; else if ((nextIndex = transferIndex) <= 0) { i = -1; advance = false; } // 用CAS计算得到的transferIndex 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; // table 指向nextTable sizeCtl = (n << 1) - (n >>> 1); // sizeCtl阈值为原来的1.5倍 return; // 跳出死循环， } // CAS 更扩容阈值，在这里面sizectl值减一，说明新加入一个线程参与到扩容操作 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 } } // 遍历的节点为null，则放入到ForwardingNode 指针节点 else if ((f = tabAt(tab, i)) == null) advance = casTabAt(tab, i, null, fwd); // f.hash == -1 表示遍历到了ForwardingNode节点，意味着该节点已经处理过了 // 这里是控制并发扩容的核心 else if ((fh = f.hash) == MOVED) advance = true; // already processed else { // 节点加锁 synchronized (f) { // 节点复制工作 if (tabAt(tab, i) == f) { Node<K,V> ln, hn; // fh >= 0 ,表示为链表节点 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); } // 在nextTable i 位置处插上链表 setTabAt(nextTab, i, ln); // 在nextTable i + n 位置处插上链表 setTabAt(nextTab, i + n, hn); // 在table i 位置处插上ForwardingNode 表示该节点已经处理过了 setTabAt(tab, i, fwd); // advance = true 可以执行--i动作，遍历节点 advance = true; } // 如果是TreeBin，则按照红黑树进行处理，处理逻辑与上面一致 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; } } // 扩容后树节点个数若<=6，将树转链表 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; } } } } } }扩容过程有点复杂，可以查看上面注释。这里主要涉及到多线程并发扩容,ForwardingNode的作用就是支持扩容操作，将已处理的节点和空节点置为ForwardingNode，并发处理时多个线程经过ForwardingNode就表示已经遍历了，就往后遍历，下图是多线程合作扩容的过程：
5.treeifyBin()链表转红黑树的过程
private final void treeifyBin(Node<K,V>[] tab, int index) { Node<K,V> b; int n, sc; if (tab != null) { //如果整个table的数量小于64，就扩容至原来的一倍，不转红黑树了 //因为这个阈值扩容可以减少hash冲突，不必要去转红黑树 if ((n = tab.length) < MIN_TREEIFY_CAPACITY) tryPresize(n << 1); 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 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; } //通过TreeBin对象对TreeNode转换成红黑树 setTabAt(tab, index, new TreeBin<K,V>(hd)); } } } } }addCount()方法计算ConcurrentHashMap的size
private final void addCount(long x, int check) { CounterCell[] as; long b, s; //更新baseCount，table的数量，counterCells表示元素个数的变化 if ((as = counterCells) != null || !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) { CounterCell a; long v; int m; boolean uncontended = true; //如果多个线程都在执行，则CAS失败，执行fullAddCount，全部加入count 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(); } //check>=0表示需要进行扩容操作 if (check >= 0) { Node<K,V>[] tab, nt; int n, sc; while (s >= (long)(sc = sizeCtl) && (tab = table) != null && (n = tab.length) < MAXIMUM_CAPACITY) { int rs = resizeStamp(n); if (sc < 0) { 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); } //当前线程发起库哦哦让操作，nextTable=null else if (U.compareAndSwapInt(this, SIZECTL, sc, (rs << RESIZE_STAMP_SHIFT) + 2)) transfer(tab, null); s = sumCount(); } } }put的流程现在已经分析完了，你可以从中发现，他在并发处理中使用的是乐观锁，当有冲突的时候才进行并发处理，而且流程步骤很清晰，但是细节设计的很复杂，毕竟多线程的场景也复杂.
concurrentHashMap的get操作的流程很简单，可以分为三个步骤来描述:
1.计算hash值，定位到该table索引位置，如果是首节点符合就返回。 2.如果遇到扩容的时候，会调用标志正在扩容节点ForwardingNode的find方法，查找该节点，匹配就返回。 3.以上都不符合的话，就往下遍历节点，匹配就返回，否则最后就返回null
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) {//读取首节点的Node元素 if ((eh = e.hash) == h) { //如果该节点就是首节点就返回 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; while ((e = e.next) != null) {//既不是首节点也不是ForwardingNode，那就往下遍历 if (e.hash == h && ((ek = e.key) == key || (ek != null && key.equals(ek)))) return e.val; } } return null; }
对于size的计算，在扩容和addCount()方法就已经有处理了，可以注意一下Put函数，里面就有addCount()函数，早就计算好的，然后你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) { for (int i = 0; i < as.length; ++i) { if ((a = as[i]) != null) sum += a.value; } } return sum; }HashMap线程不安全，数组+链表+红黑树 Hashtable线程安全，锁住整个对象，数组+链表 ConccurentHashMap线程安全，CAS+同步锁，数组+链表+红黑树 HashMap的key，value均可为null，其他两个不行。
在JDK1.8主要设计上的改进有以下几点:
1、不采用segment而采用node，锁住node来实现减小锁粒度。 2、设计了MOVED状态 当resize的中过程中 线程2还在put数据，线程2会帮助resize。 3、使用3个CAS操作来确保node的一些操作的原子性，这种方式代替了锁。 4、sizeCtl的不同值来代表不同含义，起到了控制的作用。采用synchronized而不是ReentrantLock
