简介

由于HashMap是非线程安全的，而且HashTable和Collections.synchronizedMap()的效率很低（基本上是对读写操作加锁，一个线程在使用，其他线程必须等待）。因此可以使用并发安全的ConcurrentHashMap。

ConcurrentHashMap的实现原理和HashMap有很多相似之处，所以看了HashMap的源码后对于理解ConcurrentHashMap有很大的好处。

JDK1.7

ConcurrentHashMap采用 分段锁(Segments) 的机制，底层采用数组+链表的存储结构。

Segments继承了 ReentrantLock 用来充当锁的角色，每个Segment保护哈希表(table[])的若干个桶(HashBucket)。

JDK1.8

JDK1.8已经不使用分段锁机制来保证并发安全了，而是使用 CAS+Synchronized 来保证，底层使用数组+链表+红黑树的存储结构（类似于HashMap的改变）。

重要属性

以下是一些会用到的属性，部分在HashMap已经出现过。

其中比较重要的是 sizeCtl，这个标志控制了很多状态。

/ 最大容量
private static final int MAXIMUM_CAPACITY = 1 << 30;// 默认初始容量
private static final int DEFAULT_CAPACITY = 16;// 并发级别,主要是为了兼容之前的版本
private static final int DEFAULT_CONCURRENCY_LEVEL = 16;// 负载因子
private static final float LOAD_FACTOR = 0.75f;/*** The number of bits used for generation stamp in sizeCtl.* Must be at least 6 for 32bit arrays.*/
private static int RESIZE_STAMP_BITS = 16;/*** The maximum number of threads that can help resize.* Must fit in 32 - RESIZE_STAMP_BITS bits.*/
private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1;/*** The bit shift for recording size stamp in sizeCtl.*/
private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;/** Encodings for Node hash fields. See above for explanation.* 某些结点的hash值,在之后用这些去判断某个结点的类型*/
static final int MOVED     = -1; // hash for forwarding nodes
static final int TREEBIN   = -2; // hash for roots of trees
static final int RESERVED  = -3; // hash for transient reservations
static final int HASH_BITS = 0x7fffffff; // usable bits of normal node hash// 哈希表数组,在第一次插入的时候才初始化,大小是2的幂,有volatile修饰
transient volatile Node<K,V>[] table;// 扩容时使用,用来取代旧的table数组,有volatile修饰
private transient volatile Node<K,V>[] nextTable;// 记录容器容量大小
private transient volatile long baseCount;// -1是在初始化,-n表示有(n-1)个线程在扩容,等于0为默认值，大于0表示扩容阈值
private transient volatile int sizeCtl;// 需要遍历的下标
private transient volatile int transferIndex;/*** Spinlock (locked via CAS) used when resizing and/or creating CounterCells.*/
private transient volatile int cellsBusy;// 在高并发时候把对单个值的更新转化为数组上的更新，降低并发争用
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类

这里的Node类不允许直接setValue()，并且val和next使用了volatile修饰保证了可见性。

static class Node<K,V> implements Map.Entry<K,V> {final int hash;final K key;// 使用volatile保证可见性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; }// 不允许直接setValuepublic 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;}
}

ForwardingNode类

这个类对于ConcurrentHashMap很重要，是实现并发的核心之一。

这个类是用来标识table[]上的Node的，当表上的结点是 ForwardingNode 类时，说明这个结点已经被复制了，不需要再对这个结点进行操作了。

在后面很多方法中都有这个类的出现。

// ForwardingNode类
static final class ForwardingNode<K,V> extends Node<K,V> {final Node<K,V>[] nextTable;ForwardingNode(Node<K,V>[] tab) {// 标志结点hash值为MOVED(-1)super(MOVED, null, null, null);this.nextTable = tab;}// 从nextTable中查询结点Node<K,V> find(int h, Object k) { ... }
}

原子操作和Unsafe类

这里有三个重要的原子操作，使用这些操作而不需要加锁保证了ConcurrentHashMap的性能。

/***  三个重要的原子操作*  ((long)i << ASHIFT) + ABASE 用来计算在内存中的偏移量*  ASHIFT是指tab[i]中第i个元素在相对于数组第一个元素的偏移量，而ABASE是数组的第一个位置的元素在内存中的偏移地址*/
// 获取i处Node,即tab[i]
static final <K,V> ConcurrentHashMap.Node<K,V> tabAt(ConcurrentHashMap.Node<K,V>[] tab, int i) {return (ConcurrentHashMap.Node<K,V>)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE);
}
// 利用CAS算法设置i位置上的Node节点（将c和tab[i]比较，相同则插入v）
static final <K,V> boolean casTabAt(ConcurrentHashMap.Node<K,V>[] tab, int i,ConcurrentHashMap.Node<K,V> c, ConcurrentHashMap.Node<K,V> v) {// CAS算法:无阻塞，通过自旋来实现不断比较期望值与当前值，若相等，则修改，否则一直自旋（与乐观锁思想相似）return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v);
}
// 设置节点位置的值
static final <K,V> void setTabAt(ConcurrentHashMap.Node<K,V>[] tab, int i, ConcurrentHashMap.Node<K,V> v) {U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v);
}

Unsafe类提供了很多操作。例如获取元素的地址等和各种CAS操作。

可以看下 Unsafe介绍

// Unsafe,U
private static final sun.misc.Unsafe U;
private static final long SIZECTL;
private static final long TRANSFERINDEX;
private static final long BASECOUNT;
private static final long CELLSBUSY;
private static final long CELLVALUE;
private static final long ABASE;
private static final int ASHIFT;  static {  try {  U = sun.misc.Unsafe.getUnsafe();  Class<?> k = ConcurrentHashMap.class;  // 获取ConcurrentHashMap这个对象字段sizeCtl在内存中的偏移量SIZECTL = U.objectFieldOffset  (k.getDeclaredField("sizeCtl"));  TRANSFERINDEX = U.objectFieldOffset  (k.getDeclaredField("transferIndex"));  BASECOUNT = U.objectFieldOffset  (k.getDeclaredField("baseCount"));  CELLSBUSY = U.objectFieldOffset  (k.getDeclaredField("cellsBusy"));  Class<?> ck = CounterCell.class;  CELLVALUE = U.objectFieldOffset  (ck.getDeclaredField("value")); Class<?> ak = Node[].class;  // 获取数组第一个元素的偏移地址ABASE = U.arrayBaseOffset(ak); // arrayIndexScale可以获取数组的转换因子，也就是数组中元素的增量地址 int scale = U.arrayIndexScale(ak);  if ((scale & (scale - 1)) != 0)  throw new Error("data type scale not a power of two");  ASHIFT = 31 - Integer.numberOfLeadingZeros(scale);  } catch (Exception e) {  throw new Error(e);  }
}  

重要方法

初始化表操作(initTable)

这个方法的目的是初始化一个table。

// 初始化表
private final ConcurrentHashMap.Node<K,V>[] initTable() {ConcurrentHashMap.Node<K,V>[] tab; int sc;while ((tab = table) == null || tab.length == 0) {// 如果已经创建过了则让行if ((sc = sizeCtl) < 0)Thread.yield(); // lost initialization race; just spin// CAS操作,若SIZECTL和sc相同,则将SIZECTL修改为-1(表示正在初始化)else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {try {if ((tab = table) == null || tab.length == 0) {// 创建表int n = (sc > 0) ? sc : DEFAULT_CAPACITY;@SuppressWarnings("unchecked")ConcurrentHashMap.Node<K,V>[] nt = (ConcurrentHashMap.Node<K,V>[])new ConcurrentHashMap.Node<?,?>[n];table = tab = nt;//相当于sc=0.75*n 设置一个扩容的阈值sc = n - (n >>> 2);}} finally {sizeCtl = sc;}break;}}return tab;
}

插入键值对(put和putVal)

put操作和putVal的操作的关系只是一个调用关系，在这就不提put操作了，重点在于putVal。

这里的操作流程是：

• 先计算传入key的哈希值hash。
• 进入for循环自旋直到完成插入操作。
  1. 如果表还未初始化，则去初始化表。
  2. 如果hash位置对应的桶还未初始化，就用CAS操作去插入新的键值对并退出自旋。
  3. 如果是ForwardingNode就调用 helpTransfer() 去帮忙将旧表复制到新表中。
  4. 否则就是需要将新的键值对放到链表或者树上了。（具体看代码）
• 最后调用 addCount() 使得元素数目+1，这里如果不够空间也会在 addCount() 中扩容。

final V putVal(K key, V value, boolean onlyIfAbsent) {// 不允许key或value为null,这里和HashMap不一样if (key == null || value == null) throw new NullPointerException();/** static final int spread(int h) {*     return (h ^ (h >>> 16)) & HASH_BITS;* }* static final int HASH_BITS = 0x7fffffff; // usable bits of normal node hash* 计算哈希*/int hash = spread(key.hashCode());int binCount = 0;// 自旋操作,只有成功插入了才会跳出for (ConcurrentHashMap.Node<K,V>[] tab = table;;) {ConcurrentHashMap.Node<K,V> f; int n, i, fh;// 表还未初始化if (tab == null || (n = tab.length) == 0)tab = initTable();// 该位置还没有初始化else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {// 使用CAS插入新的键值对,不需要加锁if (casTabAt(tab, i, null,new ConcurrentHashMap.Node<K,V>(hash, key, value, null)))break;                   // no lock when adding to empty bin}// 如果是Forwording Node就帮忙transfer整合表else if ((fh = f.hash) == MOVED)tab = helpTransfer(tab, f);else {V oldVal = null;// 这里使用synchronized同步table中目标位置的bucket,即tab[i],相当于分段锁,保证线程安全synchronized (f) {if (tabAt(tab, i) == f) {// 如果是链表结点if (fh >= 0) {binCount = 1;// 遍历链表for (ConcurrentHashMap.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;}ConcurrentHashMap.Node<K,V> pred = e;// 找到链表末尾还找不到就在末尾插入新的键值对if ((e = e.next) == null) {pred.next = new ConcurrentHashMap.Node<K,V>(hash, key,value, null);break;}}}// 如果是树的结点else if (f instanceof ConcurrentHashMap.TreeBin) {ConcurrentHashMap.Node<K,V> p;binCount = 2;// 树的操作if ((p = ((ConcurrentHashMap.TreeBin<K,V>)f).putTreeVal(hash, key,value)) != null) {oldVal = p.val;if (!onlyIfAbsent)p.val = value;}}}}// 为1表示为链表结点，为2表示为树结点，为0表示插入新的结点if (binCount != 0) {// 判断是否达到变成树的阈值(默认为8),达到了就将tab[i]的链表重构为红黑树if (binCount >= TREEIFY_THRESHOLD)treeifyBin(tab, i);if (oldVal != null)return oldVal;break;}}}// 当前map的元素+1addCount(1L, binCount);return null;
}

helpTransfer

这个方法是发现结点是 ForwardingNode 类时候调用的，进入其中帮助 transfer。

/*** Helps transfer if a resize is in progress.* 辅助方法,f为ForwardingNode进入,细节个人还没理解透彻*/
final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {Node<K,V>[] nextTab; int sc;if (tab != null && (f instanceof ForwardingNode) &&(nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) {// 返回一个扩容校验标识int rs = resizeStamp(tab.length);// 当处于扩容状态while (nextTab == nextTable && table == tab &&(sc = sizeCtl) < 0) {// 判断校验标识是否相等，如果校验符不等或者扩容操作已经完成了，直接退出循环，不用协助它们扩容了if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||sc == rs + MAX_RESIZERS || transferIndex <= 0)break;// 否则调用transfer帮助它们进行扩容,sc+1标识增加了一个线程进行扩容if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {transfer(tab, nextTab);break;}}return nextTab;}return table;
}

扩容操作（transfer）

当空间不够的时候，要将旧的表(table)复制到新的表(nextTab)中。这是个人觉得最复杂的操作了。

操作流程：

• 如果新的表为空，就初始化新的表（单线程操作），新的表的容量为原来的2倍。
• 新建一个 ForwardingNode 用来标志已经完成扩容的结点。
• 自旋直到处理完毕
  • 为线程分配工作的区间，逆序处理每一个桶。
  • 遍历桶。
  • 如果该桶为空，就将 ForwardingNode 放入标志已经处理过。
  • 如果该桶为 ForwardingNode 就跳过。
  • 否则使用 synchronized 锁住桶，进行复制转移操作（详见注释，和HashMap有点像）。完成后，标记为 ForwardingNode。
  • 当finish后，则将原来的table更新为nextTab，然后将nextTable设为null帮助GC。

/*** Moves and/or copies the nodes in each bin to new table. See* above for explanation.* 扩容，将旧的表的元素放到新的表中*/
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) {            // initiatingtry {// 扩展为原容量的2倍(n<<1),这个操作是单线程完成的,因为这初始化了nextTab@SuppressWarnings("unchecked")Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];nextTab = nt;} catch (Throwable ex) {      // try to cope with OOMEsizeCtl = Integer.MAX_VALUE;return;}nextTable = nextTab;// transferIndex指向最后一个桶,从后往前遍历transferIndex = n;}// 下面是并发扩容的核心int nextn = nextTab.length;// 新建一个ForwardingNode,用于标识已经完成复制转移的桶ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);// 如果等于true,那么说明这个节点已经处理过boolean advance = true;// 标识是否所有结点复制完成boolean finishing = false; // to ensure sweep before committing nextTab// i是当前处理的桶,bound是区间下界for (int i = 0, bound = 0;;) {Node<K,V> f; int fh;while (advance) {int nextIndex, nextBound;// 通过 --i 去遍历桶if (--i >= bound || finishing)advance = false;// 说明已经没有需要复制转移的桶了else if ((nextIndex = transferIndex) <= 0) {i = -1;advance = false;}// 为当前线程分配任务，处理的桶结点区间为（nextBound,nextIndex）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;// 扩容阈值为原来的1.5倍,即现在的0.75倍sizeCtl = (n << 1) - (n >>> 1);return;}// CAS将sizeCtl-1,说明新的线程加入到扩容操作中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}}// 如果是空结点,就将ForwardingNode放入用来标志已经被处理过else if ((f = tabAt(tab, i)) == null)advance = casTabAt(tab, i, null, fwd);// 如果是ForwardingNode,那么这个点已经被处理过,跳过else if ((fh = f.hash) == MOVED)advance = true; // already processedelse {// 否则进入转移synchronized (f) {if (tabAt(tab, i) == f) {Node<K,V> ln, hn;// 是一个链表结点if (fh >= 0) {// 这里和HashMap一样,只要判断扩容后的那一位是1还是0就可以知道是放在原位i还是放到i+n的桶int runBit = fh & n;Node<K,V> lastRun = f;// 遍历找到桶中最后连续的 fh&n 不变的结点for (Node<K,V> p = f.next; p != null; p = p.next) {int b = p.hash & n;if (b != runBit) {runBit = b;lastRun = p;}}// 如果是0,那么就让链表头为ln(放到原位i的链表)if (runBit == 0) {ln = lastRun;hn = null;} // 如果是1,那么让链表头为hn(放到i+n的链表)else {hn = lastRun;ln = null;}/*** 如果是ln一开始有值的话,那么就是lastRun后面那一段是顺序的,然后其余的一个个插入到lastRun前面* 如果是hn一开始有值的话,也是同理*/for (Node<K,V> p = f; p != lastRun; p = p.next) {int ph = p.hash; K pk = p.key; V pv = p.val;/***  Node(int hash, K key, V val, Node<K,V> next) {*      this.hash = hash;*      this.key = key;*      this.val = val;*      this.next = next;*  }*  这里Node的构造函数是让新的结点的next指向传入的结点*  即会让链表逆序*/if ((ph & n) == 0)ln = new Node<K,V>(ph, pk, pv, ln);elsehn = new Node<K,V>(ph, pk, pv, hn);}// 将两条链表放入到nextTab的相应的桶中setTabAt(nextTab, i, ln);setTabAt(nextTab, i + n, hn);// 将原来的桶的第i位标识为已经处理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;elseloTail.next = p;loTail = p;++lc;}else {if ((p.prev = hiTail) == null)hi = p;elsehiTail.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，更新baseCount并判断table数组是否太小需要扩容。

核心思想是在并发较低时，只需更新base值。在高并发的情况下，将对单一值的更新转化为数组上元素的更新，以降低并发争用。总的映射个数为base＋CounterCell各个元素的和。如果总数大于阈值，扩容。

private final void addCount(long x, int check) {// 初始化时counterCells为空，在并发量很高时，如果存在两个线程同时执行CAS修改baseCount值，// 则失败的线程会继续执行方法体中的逻辑，使用CounterCell记录元素个数的变化CounterCell[] as; long b, s;// 更新baseCount为sif ((as = counterCells) != null ||!U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {// 更新失败则进入CounterCell a; long v; int m;// uncontended表示更新CounterCell是否存在争用boolean uncontended = true;// CAS更新CounterCell数组的值+xif (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添加到counterCells[]或baseCount中fullAddCount(x, uncontended);return;}if (check <= 1)return;// 统计元素个数s = sumCount();}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) {// 扩容结束或者没有桶分配就breakif ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||transferIndex <= 0)break;// CAS更新正在扩容的线程+1if (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);// 实时监测s = sumCount();}}
}

总结

• JDK1.8使用了CAS操作和synchronized来实现并发，但是这里的synchronized只是锁住正在操作的桶而已，并没有锁住整个map。而JDK1.7使用了分段锁来实现并发，一个segment对应了多个桶。
• JDK1.8使用链表+红黑树来实现，在冲突很多的情况下时间复杂度优化了许多。
• 阅读了HashMap源码后再看ConcurrentHashMap简单了一些，两者有很多共通之处。

Reference

• http://www.cnblogs.com/yangming1996/p/8031199.html#commentform
• http://www.cnblogs.com/huaizuo/p/5413069.html