Java之集合(二十五)ConcurrentHashMap
转载请注明源出处:http://www.cnblogs.com/lighten/p/7520808.html
1.前言
本章介绍使用的最频繁的并发集合类之一ConcurrentHashMap,之前介绍过HashMap和HashTable,指出了HashTable的问题。虽然可以使用Collections.synchronizedMap方法包装HashMap完成线程安全的Map,但是这样做是远远无法满足我们对性能的需求。因为Map使用频率十分高,键值对在程序中是十分常见的,每一次put或get操作锁住整个Map,无疑是十分糟糕的。值得一提的是ConcurrentHashMap在早起的版本中好像是采取了分段式锁的方法进行控制的,但是至少JDK7之后(JDK8没有使用segment的方式,只装了7,8)并没有采取这种方式,而是使用CAS操作与synchronized锁更快的方法实现的,Segment的存在意义也只是为了兼容之前的版本,但是实际还是一种分段锁的思路,锁的是hash桶。
通常检索操作(包括get)都是非阻塞的,所以可能与更新操作重叠(put,remove)。检索结果受到最近的更新操作的影响,简单说就是一个键的更新操作对于任何检索操作而已表现成happens-before的关系,所以键会展示出最近更新的值。对于总的操作(putAll,clear)。检索可能受到部分键值对的插入或移除影响。同样的,迭代器受到在创建的时候某个时刻hash表的状态影响,并不会抛出并发异常,但是迭代器的设计只考虑了单线程的使用。
冲突严重的时候Hash表会自动扩容,扩容的遍历操作被视为是一个缓慢的操作,如果可能通过初始容量构造参数设置大小,loadFactor和HashMap一样默认是0.75。
注意ConcurrentHashMap是不允许键或值为null,HashMap可以。
2.ConcurrentHashMap
2.1 数据结构
ConcurrentHashMap有许多的参数,上图是部分变量参数。还有很多常量参数:
MAXIMUM_CAPACITY:最大容量230。
DEFAULT_CAPACITY:默认容量16
MAX_ARRAY_SIZE:最大数组大小(toArray等方法使用)Integer.MAX_VALUE – 8
DEFAULT_CONCURRENCY_LEVEL:默认并发级别16(无用,兼容前版本)
LOAD_FACTOR:载入因子0.75
TREEIFY_THRESHOLD:转变红黑树的阈值8
UNTREEIFY_THRESHOLD:普通链表的阈值6
MIN_TREEIFY_CAPACITY:最小的表容量,只有超过这个容量链表才可能转换成红黑树。至少要4倍的TREEIFY_THRESHOLD,默认64
MIN_TRANSFER_STRIDE:每次转移步骤最小的重建桶数量
其它的参数:
table:基本的hash表,大小是2n
nextTable:resize的时候使用
baseCount:基础的统计值,通常在无竞争的时候使用
sizeCtl:表初始化和resize的控制
transferIndex:resize时下一个表分割的索引+1
cellsBusy:resize或创建CounterCells的自旋锁
counterCells:counter cells表,非空时大小是2n
2.2 基本操作
获取一个元素,无锁:
步骤很简单:
1.通过键的hashCode计算出该键在Hash表上散列的位置h
2.如果hash表不存在或者表为空或者hash表指向h的位置(当然是通过h进一步计算定位该位置)为空,意味着没有该键的值,返回null
3.步骤2不成立,就查找该位置:
第一个的元素的hash值就相同,再比较键的值是否相等,相等就返回该值,不等就继续最后的while循环查找hash值相等,键相等的,返回其值。这里键为null就会出现空指针异常了。
第一个元素的hash值小于0,也是通过Node类自身的方法遍历链表,和后面while步骤区别不大(不明白为什么要多写这步)
存入一个元素:
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) { 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; }
存入的步骤就相对比较复杂,会不断的循环尝试。步骤如下:
1.表不存在创建表:
创建表的步骤也是一个死循环,通过sizeCtl来进行多线程初始化表控制,当有线程优先获取修改权时就会将这个值改成-1,其它线程就不能修改让出CPU执行权,继续循环尝试获取修改权(前提是还满足while循环的条件,表为null或容量为0)。获取修改权的表会使用默认的容量16创建hash表,sc会设置成当前容量的3/4,最后将其赋值给sizeCtl。
2.如果表存在就会查找该键所在表的位置,该位置还没有值意味着无冲突,首次设置,就通过CAS尝试设置,设置成功跳出循环,失败了也不要紧,继续循环。
3.如果指定位置第一个元素的hash是被标记成moved状态,进行transfer操作调整后再循环,transfer具体操作如下:
看方法上的注释也能够明白moved状态意味着该位置的所有节点处于resize的移动状态,这里处理的就是扩容和put之间的冲突了。表不为null且节点是ForwardingNode且nextTable不为空,判断是在resize,后面的计算原理不太清楚,但还是能明白这段代码所起到的作用。先计算出rs,resize的时候nextTable,table和sizeCtl都是小于0的,才是未结束,进入循环在判断一下sc和rs的关系和transferIndex下标判断有没有处理完,处理完了就跳出循环,没处理完就会抢占处理,最终都是返回转移的nextTab,put方法就在resize阶段无缝衔接到新的表上了。这里逻辑很清楚,并发put的时候helpTransfer也可能并发调用,我们预期的应该是返回一个新表,但是新表的while循环是根据sizeCtl来控制的,意味着sizeCtl<0的时候移动已经完成,返回新表是没什么问题的,所以这个值是最后才会改变,其它的情况所有的线程都会获取新表,而更上一层的moved状态应该是更后面改变,这样就不会产生问题,所有新加的元素都进入更新后的表。而循环中满足跳出条件就意味着转换完成,这些变化发生的应该更早。最大的疑问在于抢夺transfer权限之后sc的值改变了,后续会直接返回nextTab,这个时间nextTable真的准备好了吗,也就是transfer的操作可能慢了,导致实际nextTab还没有准备好。否则sc+1还是小于0的,依旧进入了循环,但是多个线程可能同时操作tranfer方法,可能会造成处理冲突。所有的关键在于transfer方法的实现了。
transfer的代码过长,这里简述一下其步骤:
1.计算stride的值,单核就是表长度,多核是表长度/8/核数,最小16,这个字段的含义是一次处理多少个hash桶。
2.如果nextTab==null,创建nextTab,原长度的2倍
3.构建ForwardingNode节点,其hash就是模式moved,里面包含的就是新表
4.死循环开始转移节点到新表:
先循环找到下一个处理的hash表位置。过程是倒着处理的,先处理数组的后面部分,transferIndex指示下一次处理的下标开始位置。stride是一次处理多少个hash桶。跳过这个查找位置的方法就是所有节点都已经被分配处理了,通过CAS线程抢夺处理权,这里就巧妙的进行了多线程分段处理,各个线程互不干扰。只要其他线程要等所有处理完才结束,那么就不会产生线程安全问题。
如果i<0等条件成立,意味着当前线程没有可以处理的,判断sc的大小,如果符合结束的判断改变并设置finish,在下一个循环完成表的更新。这里可以看出结束判断都是交给sc的变化,这个对所有线程都是一样处理判断的,判断不通过会直接返回,任何线程进入resize的transfer环节都会使得sc+1,所有线程离开都会-1,问题是只有一个线程才会知道所有的都处理完成了,改变table。(这导致了上述问题,put在resize环节,返回的table可能是没有准备好的newTable,但是并不要紧,因为转移的过程中锁住了表的hash桶第一个元素,put操作锁的是同一个,所以依旧要等转换完成)
i是当前线程处理的hash表下标,如果为null,设置成ForwardingNode,即Moved模式
不为null,但是hash是moved状态,表明正在处理,需要重新选择处理的位置。
锁住hash表下标对象f,如果hash表下标f没有改变才意味着没有变化,能进行转移。转移的部分就不再进行描述,上面选择处理端,锁住处理的下标就足够保证线程安全了。顺便一提,其在处理的时候就会将第一个元素改成ForwardingNode,put的时候就知道这个正在处理,会进入helpTransfer。
4.锁住当前表下标第一个元素,进行插入操作。这里有个问题,虽然和resize环节一样锁的是表下标的第一个元素,但是会不会出现不一致的情况?transfer的f计算是通过旧表的得到的第一个,put环节第一回合确实是旧表,但是helpTransfer之后就会变成新表,再循环的时候不是出现了不一致了吗,此时新表还是未准备好的表。实际上不会,因为只有该下标处理完了,才会设置成moved状态,put环节才会切换成新表,否则就会进入锁的环节,此时如果正在转移这个区间的内容,就不会进入,没转移也不要紧,直接放入旧表就好了,之后处理的时候会一并处理。而synchronize在put等待transfer完成之后,做了再次判断第一个节点是否是一致了,就不用担心会插入旧表,因为新表完成后旧表该节点就是ForwardingNode节点,不会相等,下次循环就进入切换到新表了。
5.计数,不进行描述,通过CounterCell。
其它的方法也就不一一叙述了,总的来说检索方法没有加锁,插入移除方法考虑了扩容的问题,通过巧妙的利用CAS和锁住下标的第一个对象完成,moved状态都是转换完成的hash表,获取新表直接插入相关位置就可以了,未完成的,直接插入旧表等待完成转移就可以了。其实这里还会产生一个问题,旧表插入的位置和新表hash出来的是同一个位置吗?我想答案肯定是不是相同的,那么会导致一个新问题,旧表的hash桶X都移动了新表的各个位置,但是put阻塞后变成新表,该键值hash出来的不一样了,会落到不同的新表位置,再想一想,如果新表散布了旧表的元素,这样转移是会有问题的。所以实际上hash的方法和一般的hash不一样,而且转移的时候是将旧表扩容了两倍,旧表的桶也是拆成了两个,应该是用了某种算法确保这个桶的元素只会落在这两个新桶上吧,这样就确保了旧表新加的元素也不会落到新表的其它桶上,完美的实现了桶的整体移动而不会打散元素,造成整个结构不可用。具体是如何计算的,计算原理没有研究。