为什么要再写一篇ConcurrentHashMap的文章,有下面几个原因:
1. jdk1.8 和我上次写的1.6版本的在实现上差距很大,我也是今天看了下才发现,去年又一次去面试刚好问道这个地方了,我就胸有成竹的回答了有关同步,锁,效率的问题,今天一看基本全错了。
2. 上次写的文章有点笼统,没有触及到问题的根本,只是在代码层面走了流程,这次我需要把没完成的问题一一解决。
所以此文会解决如下几个问题:
1. 基本结构是什么样的,关键的操作(增加,修改,删除会如何影响结构)
2. 既然遍历可以不像HashTable一样出异常,那么它是怎么做到的
3. 同样比HashTable快,又快在哪里,为什么会快呢
一、基本结构&基本操作
我们都知道HashMap的数据结构是数组+链表的结构,我之前认为高端一点的说法应该是邻接表的结构,但是我上网搜了下很少有人直接说是邻接表。
//数组 /** * The array of bins. Lazily initialized upon first insertion. * Size is always a power of two. Accessed directly by iterators. */ transient volatile Node<K,V>[] table; /** * The next table to use; non-null only while resizing. */ private transient volatile Node<K,V>[] nextTable; //链表 static class Node<K,V> implements Map.Entry<K,V> { final int hash; final K key; volatile V val; volatile Node<K,V> next; ...... }
这里会有两个数组的情况,为什么会这样呢,注释上说的清楚,当扩张大小的时候会用到这张表,比如说之前有16个元素的数组,当第16 * LOAD_FACTOR +1 = 16* 0.75 +1 = 12 +1 = 13的时候,会新建一个数组大小为16 << 1 = 32 ,同样下一次的变动大小为 25。如果你想看jdk中是如何变动的,可以看下面的代码
//方法调用为 putVal() -> addCount() -> transfer() 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];//看这里 ...... }
所以添加的时候,会根据key的hash值,映射到当前数组的某一个元素也就是Node,然后在从当前Node找key为新添加元素的k,如果等于表示已经存在key,进行更新操作,否则进行添加操作。添加完之后在看size()是否达到扩展的情况,如果达到了执行上面扩大数据的操作了。这里注意一下,新版本的hash映射到数组元素的算法已经修改了。
/** * Spreads (XORs) higher bits of hash to lower and also forces top * bit to 0. Because the table uses power-of-two masking, sets of * hashes that vary only in bits above the current mask will * always collide. (Among known examples are sets of Float keys * holding consecutive whole numbers in small tables.) So we * apply a transform that spreads the impact of higher bits * downward. There is a tradeoff between speed, utility, and * quality of bit-spreading. Because many common sets of hashes * are already reasonably distributed (so don't benefit from * spreading), and because we use trees to handle large sets of * collisions in bins, we just XOR some shifted bits in the * cheapest possible way to reduce systematic lossage, as well as * to incorporate impact of the highest bits that would otherwise * never be used in index calculations because of table bounds. */ static final int spread(int h) { return (h ^ (h >>> 16)) & HASH_BITS; } 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(); else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) { //tabAt 也是新的 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; synchronized (f) { if (tabAt(tab, i) == f) { 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; } } } 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) { if (binCount >= TREEIFY_THRESHOLD) treeifyBin(tab, i); if (oldVal != null) return oldVal; break; } } } addCount(1L, binCount); return null; }
tabAt 是根据hash 定位的方法,源代码如下
/* * Volatile access methods are used for table elements as well as * elements of in-progress next table while resizing. All uses of * the tab arguments must be null checked by callers. All callers * also paranoically precheck that tab's length is not zero (or an * equivalent check), thus ensuring that any index argument taking * the form of a hash value anded with (length - 1) is a valid * index. Note that, to be correct wrt arbitrary concurrency * errors by users, these checks must operate on local variables, * which accounts for some odd-looking inline assignments below. * Note that calls to setTabAt always occur within locked regions, * and so in principle require only release ordering, not * full volatile semantics, but are currently coded as volatile * writes to be conservative. */ @SuppressWarnings("unchecked") static final <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i) { return (Node<K,V>)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE); }
U 是一个sun.misc.Unsafe 的实例,Unsafe是一个低级别的,不安全的方法集合,主要用在一些多线程处理方面的优化,绝大部分是native的方法。
二、安全的遍历如何做到
说到遍历我们用的最多的是下面这种方式了
ConcurrentHashMap<K,V> map ; map.entrySet().iterator(); while(iterator.hasNext){ iterator.next(); }
接下来我们看看,entrySet()这个方法
public Set<Map.Entry<K,V>> entrySet() { EntrySetView<K,V> es; return (es = entrySet) != null ? es : (entrySet = new EntrySetView<K,V>(this)); }
方法中使用的是EntrySetView这个类包装的,注意到参数this,表示把当前的Map传递过去了的,为什么传递过去呢,很简单就为了得到map中的数组和链表,也就是数据。接下来看看EntrySetView这个类
static final class EntrySetView<K,V> extends CollectionView<K,V,Map.Entry<K,V>> implements Set<Map.Entry<K,V>>, java.io.Serializable { private static final long serialVersionUID = 2249069246763182397L; EntrySetView(ConcurrentHashMap<K,V> map) { super(map); } ...... //迭代器方法如下 public Iterator<Map.Entry<K,V>> iterator() { ConcurrentHashMap<K,V> m = map; Node<K,V>[] t; int f = (t = m.table) == null ? 0 : t.length; return new EntryIterator<K,V>(t, f, 0, f, m); } ...... }
迭代器也有一个包装类来完成EntryIterator, 同样注意这个类的构造方法有5个参数(map中的节点数组,数组的长度,0,数组的长度,map本身的引用)。看看这个看似简单的类是如何工作的。
static final class EntryIterator<K,V> extends BaseIterator<K,V> implements Iterator<Map.Entry<K,V>> { EntryIterator(Node<K,V>[] tab, int index, int size, int limit, ConcurrentHashMap<K,V> map) { super(tab, index, size, limit, map); } public final Map.Entry<K,V> next() { Node<K,V> p; if ((p = next) == null) throw new NoSuchElementException(); K k = p.key; V v = p.val; lastReturned = p; advance(); return new MapEntry<K,V>(k, v, map); } }
这里只是把next node的值取出来包装到MapEntry中返回就好,具体如何找next呢,关键来了,看advance()这个方法。
/** * Advances if possible, returning next valid node, or null if none. */ final Node<K,V> advance() { Node<K,V> e; if ((e = next) != null) e = e.next; for (;;) { Node<K,V>[] t; int i, n; // must use locals in checks if (e != null) return next = e; if (baseIndex >= baseLimit || (t = tab) == null || (n = t.length) <= (i = index) || i < 0) return next = null; if ((e = tabAt(t, i)) != null && e.hash < 0) {//特殊类型的节点hash值都为负数 if (e instanceof ForwardingNode) { tab = ((ForwardingNode<K,V>)e).nextTable; e = null; pushState(t, i, n); continue; } else if (e instanceof TreeBin) e = ((TreeBin<K,V>)e).first; else e = null; } if (stack != null) recoverState(n); else if ((index = i + baseSize) >= n) index = ++baseIndex; // visit upper slots if present } }
上面代码注意下面这一部分
if (e instanceof ForwardingNode) { tab = ((ForwardingNode<K,V>)e).nextTable; e = null; pushState(t, i, n); continue; }
代码大意很明确,意思是如果当前节点Node,是ForwardingNode的类型的时候,就使用nextTable来当作当前的table, 在最前面的数据结构中我们提到了有两个table,另一个table只有当调整空间也就是resize的时候使用,此类中与resize相关的方法transfer(tab,nextTab)在最前的时候也已经讲过了,这里再多贴出点东西,看看关于ForwardingNode相关的东西。
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);//这里 ...... }
ForwardingNode表示正在移动的节点,并且此类节点的Hash值都为-1 ,
到这里大致清楚了,任何数组大小调整的时候都会有两个表,一个表应对各种外界操作,一个表应对内部大小调整,调整完全好了,才会同步到主表中。
三、快在哪里
要比较快在哪里得看在什么情况下(是否并发访问),和谁比较(HashMap,还是HashTable,这里既然说并发情况下肯定是和HashTable)比较了。
说上面一段话是想说明,任何数据类型都有一定的限制性,我很多次面试的时候问过,为什么ConcurrentHashMap要好,給我的回答是新出来的,然后反问一句,如果不好为什么要新写一个出来,弄的我也是一愣的。
说正事,通过上面的说明,ConcurrentHashMap比HashTable快的两个地方是
1. 更细粒度的锁
HashTable是通过在每个方法上加Synchronized来完成同步的,而ConcurrentHashMap是在某个Node列上(某个数组节点,里面包含了链表)加锁来实现同步。相比之下增加了锁的个数从而提高并发的数量,但同时锁的管理也更加复杂,锁的消耗也更加大,其实这也是为什么不在每一个节点(每个列上的每个子节点)上都加锁的原因。
2. 空间换时间
ConcurrentHashMap 通过增加额外的副本,来避免最耗时的resize操作,这样有两个好处一是将耗时的resize操作让其它的线程处理,二是任何时间遍历都不会出现数据不全的情况,同时不用额外锁住整个map。