Java7和8 中的 HashMap 和 ConcurrentHashMap 全解析
- 2019 年 10 月 7 日
- 筆記
Java7/8 中的 HashMap 全解析
转自https://www.javadoop.com/post/hashmap#toc7
部分内容转自
http://www.jasongj.com/java/concurrenthashmap
今天发一篇"水文"，可能很多读者都会表示不理解，不过我想把它作为并发序列文章中不可缺少的一块来介绍。本来以为花不了多少时间的，不过最终还是投入了挺多时间来完成这篇文章的。
网上关于 HashMap 和 ConcurrentHashMap 的文章确实不少，不过缺斤少两的文章比较多，所以才想自己也写一篇，把细节说清楚说透，尤其像 Java8 中的 ConcurrentHashMap，大部分文章都说不清楚。终归是希望能降低大家学习的成本，不希望大家到处找各种不是很靠谱的文章，看完一篇又一篇，可是还是模模糊糊。
阅读建议：四节基本上可以进行独立阅读，建议初学者可按照 Java7 HashMap -> Java7 ConcurrentHashMap -> Java8 HashMap -> Java8 ConcurrentHashMap 顺序进行阅读，可适当降低阅读门槛。
阅读前提：本文分析的是源码，所以至少读者要熟悉它们的接口使用，同时，对于并发，读者至少要知道 CAS、ReentrantLock、UNSAFE 操作这几个基本的知识，文中不会对这些知识进行介绍。Java8 用到了红黑树，不过本文不会进行展开，感兴趣的读者请自行查找相关资料。
- Java7 HashMap
- 数组初始化
- 计算具体数组位置
- 添加节点到链表中
- 数组扩容
- put 过程分析
- get 过程分析
- Java7 ConcurrentHashMap
- 初始化槽: ensureSegment
- 获取写入锁: scanAndLockForPut
- 扩容: rehash
- 初始化
- put 过程分析
- get 过程分析
- 并发问题分析
- Java8 HashMap
- 数组扩容
- put 过程分析
- get 过程分析
- Java8 ConcurrentHashMap
- 数据迁移：transfer
- 初始化数组：initTable
- 链表转红黑树: treeifyBin
- 初始化
- put 过程分析
- 扩容：tryPresize
- get 过程分析
- 总结
Java7 HashMap
HashMap 是最简单的，一来我们非常熟悉，二来就是它不支持并发操作，所以源码也非常简单。
首先，我们用下面这张图来介绍 HashMap 的结构。
这个仅仅是示意图，因为没有考虑到数组要扩容的情况，具体的后面再说。
大方向上，HashMap 里面是一个数组，然后数组中每个元素是一个单向链表。
上图中，每个绿色的实体是嵌套类 Entry 的实例，Entry 包含四个属性：key, value, hash 值和用于单向链表的 next。
capacity：当前数组容量，始终保持 2^n，可以扩容，扩容后数组大小为当前的 2 倍。
loadFactor：负载因子，默认为 0.75。
threshold：扩容的阈值，等于 capacity * loadFactor
put 过程分析
还是比较简单的，跟着代码走一遍吧。
public V put(K key, V value) { // 当插入第一个元素的时候，需要先初始化数组大小 if (table == EMPTY_TABLE) { inflateTable(threshold); } // 如果 key 为 null，感兴趣的可以往里看，最终会将这个 entry 放到 table[0] 中 if (key == null) return putForNullKey(value); // 1. 求 key 的 hash 值 int hash = hash(key); // 2. 找到对应的数组下标 int i = indexFor(hash, table.length); // 3. 遍历一下对应下标处的链表，看是否有重复的 key 已经存在， // 如果有，直接覆盖，put 方法返回旧值就结束了 for (Entry<K,V> e = table[i]; e != null; e = e.next) { Object k; if (e.hash == hash && ((k = e.key) == key || key.equals(k))) { V oldValue = e.value; e.value = value; e.recordAccess(this); return oldValue; } } modCount++; // 4. 不存在重复的 key，将此 entry 添加到链表中，细节后面说 addEntry(hash, key, value, i); return null; }
数组初始化
在第一个元素插入 HashMap 的时候做一次数组的初始化，就是先确定初始的数组大小，并计算数组扩容的阈值。
private void inflateTable(int toSize) { // 保证数组大小一定是 2 的 n 次方。 // 比如这样初始化：new HashMap(20)，那么处理成初始数组大小是 32 int capacity = roundUpToPowerOf2(toSize); // 计算扩容阈值：capacity * loadFactor threshold = (int) Math.min(capacity * loadFactor, MAXIMUM_CAPACITY + 1); // 算是初始化数组吧 table = new Entry[capacity]; initHashSeedAsNeeded(capacity); //ignore }
这里有一个将数组大小保持为 2 的 n 次方的做法，Java7 和 Java8 的 HashMap 和 ConcurrentHashMap 都有相应的要求，只不过实现的代码稍微有些不同，后面再看到的时候就知道了。
计算具体数组位置
这个简单，我们自己也能 YY 一个：使用 key 的 hash 值对数组长度进行取模就可以了。
static int indexFor(int hash, int length) { // assert Integer.bitCount(length) == 1 : "length must be a non-zero power of 2"; return hash & (length-1); }
这个方法很简单，简单说就是取 hash 值的低 n 位。如在数组长度为 32 的时候，其实取的就是 key 的 hash 值的低 5 位，作为它在数组中的下标位置。
添加节点到链表中
找到数组下标后，会先进行 key 判重，如果没有重复，就准备将新值放入到链表的表头。
void addEntry(int hash, K key, V value, int bucketIndex) { // 如果当前 HashMap 大小已经达到了阈值，并且新值要插入的数组位置已经有元素了，那么要扩容 if ((size >= threshold) && (null != table[bucketIndex])) { // 扩容，后面会介绍一下 resize(2 * table.length); // 扩容以后，重新计算 hash 值 hash = (null != key) ? hash(key) : 0; // 重新计算扩容后的新的下标 bucketIndex = indexFor(hash, table.length); } // 往下看 createEntry(hash, key, value, bucketIndex); } // 这个很简单，其实就是将新值放到链表的表头，然后 size++ void createEntry(int hash, K key, V value, int bucketIndex) { Entry<K,V> e = table[bucketIndex]; table[bucketIndex] = new Entry<>(hash, key, value, e); size++; }
这个方法的主要逻辑就是先判断是否需要扩容，需要的话先扩容，然后再将这个新的数据插入到扩容后的数组的相应位置处的链表的表头。
数组扩容
前面我们看到，在插入新值的时候，如果当前的 size 已经达到了阈值，并且要插入的数组位置上已经有元素，那么就会触发扩容，扩容后，数组大小为原来的 2 倍。
void resize(int newCapacity) { Entry[] oldTable = table; int oldCapacity = oldTable.length; if (oldCapacity == MAXIMUM_CAPACITY) { threshold = Integer.MAX_VALUE; return; } // 新的数组 Entry[] newTable = new Entry[newCapacity]; // 将原来数组中的值迁移到新的更大的数组中 transfer(newTable, initHashSeedAsNeeded(newCapacity)); table = newTable; threshold = (int)Math.min(newCapacity * loadFactor, MAXIMUM_CAPACITY + 1); }
这篇文章没有讲清楚1.7和1.8hashmap扩容的区别，这里特地补充：
HashMap扩容机制： JDK1.7与1.8的区别
扩容(resize)就是重新计算容量，向HashMap对象里不停的添加元素，而HashMap对象内部的数组无法装载更多的元素时，对象就需要扩大数组的长度，以便能装入更多的元素。当然Java里的数组是无法自动扩容的，方法是使用一个新的数组代替已有的容量小的数组，就像我们用一个小桶装水，如果想装更多的水，就得换大水桶。
我们分析下resize的源码，鉴于JDK1.8融入了红黑树，较复杂，为了便于理解我们仍然使用JDK1.7的代码，好理解一些，本质上区别不大，具体区别后文再说。
| 1 void resize(``int newCapacity){``//传入新的容量``2 Entry[]oldTable=table;``//引用扩容前的Entry数组``3 int oldCapacity=oldTable.length;``4 if (oldCapacity==MAXIMUM_CAPACITY){``//扩容前的数组大小如果已经达到最大(2^30)了``5 threshold=Integer.MAX_VALUE;``//修改阈值为int的最大值(2^31-1)，这样以后就不会扩容了``6 return``;``7}``8 9 Entry[]newTable=``new Entry[newCapacity];``//初始化一个新的Entry数组``10 transfer(newTable);``//！！将数据转移到新的Entry数组里``11 table=newTable;``//HashMap的table属性引用新的Entry数组``12 threshold=(``int``)(newCapacity*loadFactor);``//修改阈值``13 } |
| — |
这里就是使用一个容量更大的数组来代替已有的容量小的数组，transfer()方法将原有Entry数组的元素拷贝到新的Entry数组里。
| 1 void transfer(Entry[]newTable){``2 Entry[]src=table;``//src引用了旧的Entry数组``3 int newCapacity=newTable.length;``4 for (``int j=``0``;j<src.length;j++){``//遍历旧的Entry数组``5 Entry<K,V>e=src[j];``//取得旧Entry数组的每个元素``6 if (e!=``null``){``7 src[j]=``null``;``//释放旧Entry数组的对象引用（for循环后，旧的Entry数组不再引用任何对象）``8 do {``9 Entry<K,V>next=e.next;``10 int i=indexFor(e.hash,newCapacity);``//！！重新计算每个元素在数组中的位置``11 e.next=newTable[i];``//标记[1]``12 newTable[i]=e;``//将元素放在数组上``13 e=next;``//访问下一个Entry链上的元素``14 }``while (e!=``null``);``15 }``16 }``17 } |
| — |
newTable[i]的引用赋给了e.next，也就是使用了单链表的头插入方式，同一位置上新元素总会被放在链表的头部位置；这样先放在一个索引上的元素终会被放到Entry链的尾部(如果发生了hash冲突的话），这一点和Jdk1.8有区别，下文详解。在旧数组中同一条Entry链上的元素，通过重新计算索引位置后，有可能被放到了新数组的不同位置上。
下面举个例子说明下扩容过程。假设了我们的hash算法就是简单的用key mod 一下表的大小（也就是数组的长度）。其中的哈希桶数组table的size=2， 所以key = 3、7、5，put顺序依次为 5、7、3。在mod 2以后都冲突在table[1]这里了。这里假设负载因子 loadFactor=1，即当键值对的实际大小size 大于 table的实际大小时进行扩容。接下来的三个步骤是哈希桶数组 resize成4，然后所有的Node重新rehash的过程。
面我们讲解下JDK1.8做了哪些优化。经过观测可以发现，我们使用的是2次幂的扩展(指长度扩为原来2倍)，所以，元素的位置要么是在原位置，要么是在原位置再移动2次幂的位置。看下图可以明白这句话的意思，n为table的长度，图（a）表示扩容前的key1和key2两种key确定索引位置的示例，图（b）表示扩容后key1和key2两种key确定索引位置的示例，其中hash1是key1对应的哈希与高位运算结果。元素在重新计算hash之后，因为n变为2倍，那么n-1的mask范围在高位多1bit(红色)，因此新的index就会发生这样的变化：
此，我们在扩充HashMap的时候，不需要像JDK1.7的实现那样重新计算hash，只需要看看原来的hash值新增的那个bit是1还是0就好了，是0的话索引没变，是1的话索引变成“原索引+oldCap”，可以看看下图为16扩充为32的resize示意图：
个设计确实非常的巧妙，既省去了重新计算hash值的时间，而且同时，由于新增的1bit是0还是1可以认为是随机的，因此resize的过程，均匀的把之前的冲突的节点分散到新的bucket了。这一块就是JDK1.8新增的优化点。有一点注意区别，JDK1.7中rehash的时候，旧链表迁移新链表的时候，如果在新表的数组索引位置相同，则链表元素会倒置，但是从上图可以看出，JDK1.8不会倒置。有兴趣的同学可以研究下JDK1.8的resize源码，写的很赞，如下:
| 1 final Node<K,V>[]resize(){``2 Node<K,V>[]oldTab=table;``3int oldCap=(oldTab==``null``)?``0 :oldTab.length;``4 int oldThr=threshold;``5 int newCap,newThr=``0``;``6 if (oldCap>``0``){``7 // 超过最大值就不再扩充了，就只好随你碰撞去吧``8 if (oldCap>=MAXIMUM_CAPACITY){``9 threshold=Integer.MAX_VALUE;``10 return oldTab;``11 }``12 // 没超过最大值，就扩充为原来的2倍``13 else if ((newCap=oldCap<<``1``)<MAXIMUM_CAPACITY&&``14 oldCap>=DEFAULT_INITIAL_CAPACITY)``15 newThr=oldThr<<``1``;``// double threshold``16 }``17 else if (oldThr>``0``)``// initial capacity was placed in threshold``18 newCap=oldThr;``19else {``// zero initial threshold signifies using defaults``20 newCap=DEFAULT_INITIAL_CAPACITY;``21 newThr=(``int``)(DEFAULT_LOAD_FACTOR*DEFAULT_INITIAL_CAPACITY);``22 }``23 // 计算新的resize上限``24 if (newThr==``0``){``25``26 float ft=(``float``)newCap*loadFactor;``27 newThr=(newCap<MAXIMUM_CAPACITY&&ft<(``float``)MAXIMUM_CAPACITY?``28 (``int``)ft:Integer.MAX_VALUE);``29 }``30threshold=newThr;``31 @SuppressWarnings``({``"rawtypes"``，``"unchecked"``})``32 Node<K,V>[]newTab=(Node<K,V>[])``new Node[newCap];``33 table=newTab;``34 if (oldTab!=``null``){``35 // 把每个bucket都移动到新的buckets中``36 for (``int j=``0``;j<oldCap;++j){``37 Node<K,V>e;``38 if ((e=oldTab[j])!=``null``){``39 oldTab[j]=``null``;``40 if (e.next==``null``)``41 newTab[e.hash&(newCap-``1``)]=e;``42 else if (e``instanceof TreeNode)``43 ((TreeNode<K,V>)e).split(``this``,newTab,j,oldCap);``44 else {``// 链表优化重hash的代码块``45 Node<K,V>loHead=``null``,loTail=``null``;``46 Node<K,V>hiHead=``null``,hiTail=``null``;``47 Node<K,V>next;``48 do {``49 next=e.next;``50 // 原索引``51 if ((e.hash&oldCap)==``0``){``52if (loTail==``null``)``53 loHead=e;``54 else``55 loTail.next=e;``56 loTail=e;``57 }``58 // 原索引+oldCap``59 else {``60 if (hiTail==``null``)``61 hiHead=e;``62 else``63 hiTail.next=e;``64 hiTail=e;``65 }``66 }``while ((e=next)!=``null``);``67// 原索引放到bucket里``68 if (loTail!=``null``){``69 loTail.next=``null``;``70 newTab[j]=loHead;``71 }``72// 原索引+oldCap放到bucket里``73 if (hiTail!=``null``){``74 hiTail.next=``null``;``75 newTab[j+oldCap]=hiHead;``76 }``77 }``78}``79 }``80 }``81 return newTab;``82 } |
| — |
HashMap线程安全经典问题：并发修改时的死锁
另外这里要引入一个经典的问题，多线程put操作时可能导致的遍历死锁问题。
在多线程使用场景中，应该尽量避免使用线程不安全的HashMap，而使用线程安全的ConcurrentHashMap。那么为什么说HashMap是线程不安全的，下面举例子说明在并发的多线程使用场景中使用HashMap可能造成死循环。代码例子如下(便于理解，仍然使用JDK1.7的环境)：
| 1234567891011121314151617181920 | public class HashMapInfiniteLoop{``private static HashMap<Integer,String>map=``new HashMap<Integer,String>(``2``，``0``.75f);``public static void main(String[]args){``map.put(``5``，``"C"``);``new Thread(``"Thread1"``){``public void run(){``map.put(``7``,``"B"``);``System.out.println(map);``};``}.start();``new Thread(``"Thread2"``){``public void run(){``map.put(``3``,"A); ``System.out.println(map); ``}; ``}.start(); ``} ``} |
| — | — |
其中，map初始化为一个长度为2的数组，loadFactor=0.75，threshold=2*0.75=1，也就是说当put第二个key的时候，map就需要进行resize。
通过设置断点让线程1和线程2同时debug到transfer方法(3.3小节代码块)的首行。注意此时两个线程已经成功添加数据。放开thread1的断点至transfer方法的“Entry next = e.next;” 这一行；然后放开线程2的的断点，让线程2进行resize。结果如下图。
意，Thread1的 e 指向了key(3)，而next指向了key(7)，其在线程二rehash后，指向了线程二重组后的链表。
线程一被调度回来执行，先是执行 newTalbe[i] = e， 然后是e = next，导致了e指向了key(7)，而下一次循环的next = e.next导致了next指向了key(3)。
e.next = newTable[i] 导致 key(3).next 指向了 key(7)。注意：此时的key(7).next 已经指向了key(3)， 环形链表就这样出现了。
是，当我们用线程一调用map.get(11)时，悲剧就出现了——Infinite Loop。
get 过程分析
相对于 put 过程，get 过程是非常简单的。
- 根据 key 计算 hash 值。
- 找到相应的数组下标：hash & (length – 1)。
- 遍历该数组位置处的链表，直到找到相等(==或equals)的 key。
public V get(Object key) { // 之前说过，key 为 null 的话，会被放到 table[0]，所以只要遍历下 table[0] 处的链表就可以了 if (key == null) return getForNullKey(); // Entry<K,V> entry = getEntry(key); return null == entry ? null : entry.getValue(); }
getEntry(key):
final Entry<K,V> getEntry(Object key) { if (size == 0) { return null; } int hash = (key == null) ? 0 : hash(key); // 确定数组下标，然后从头开始遍历链表，直到找到为止 for (Entry<K,V> e = table[indexFor(hash, table.length)]; e != null; e = e.next) { Object k; if (e.hash == hash && ((k = e.key) == key || (key != null && key.equals(k)))) return e; } return null; }
Java7 ConcurrentHashMap
ConcurrentHashMap 和 HashMap 思路是差不多的，但是因为它支持并发操作，所以要复杂一些。
整个 ConcurrentHashMap 由一个个 Segment 组成，Segment 代表”部分“或”一段“的意思，所以很多地方都会将其描述为分段锁。注意，行文中，我很多地方用了“槽”来代表一个 segment。
简单理解就是，ConcurrentHashMap 是一个 Segment 数组，Segment 通过继承 ReentrantLock 来进行加锁，所以每次需要加锁的操作锁住的是一个 segment，这样只要保证每个 Segment 是线程安全的，也就实现了全局的线程安全。
concurrencyLevel：并行级别、并发数、Segment 数，怎么翻译不重要，理解它。默认是 16，也就是说 ConcurrentHashMap 有 16 个 Segments，所以理论上，这个时候，最多可以同时支持 16 个线程并发写，只要它们的操作分别分布在不同的 Segment 上。这个值可以在初始化的时候设置为其他值，但是一旦初始化以后，它是不可以扩容的。
再具体到每个 Segment 内部，其实每个 Segment 很像之前介绍的 HashMap，不过它要保证线程安全，所以处理起来要麻烦些。
初始化
initialCapacity：初始容量，这个值指的是整个 ConcurrentHashMap 的初始容量，实际操作的时候需要平均分给每个 Segment。
loadFactor：负载因子，之前我们说了，Segment 数组不可以扩容，所以这个负载因子是给每个 Segment 内部使用的。
public ConcurrentHashMap(int initialCapacity, float loadFactor, int concurrencyLevel) { if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0) throw new IllegalArgumentException(); if (concurrencyLevel > MAX_SEGMENTS) concurrencyLevel = MAX_SEGMENTS; // Find power-of-two sizes best matching arguments int sshift = 0; int ssize = 1; // 计算并行级别 ssize，因为要保持并行级别是 2 的 n 次方 while (ssize < concurrencyLevel) { ++sshift; ssize <<= 1; } // 我们这里先不要那么烧脑，用默认值，concurrencyLevel 为 16，sshift 为 4 // 那么计算出 segmentShift 为 28，segmentMask 为 15，后面会用到这两个值 this.segmentShift = 32 - sshift; this.segmentMask = ssize - 1; if (initialCapacity > MAXIMUM_CAPACITY) initialCapacity = MAXIMUM_CAPACITY; // initialCapacity 是设置整个 map 初始的大小， // 这里根据 initialCapacity 计算 Segment 数组中每个位置可以分到的大小 // 如 initialCapacity 为 64，那么每个 Segment 或称之为"槽"可以分到 4 个 int c = initialCapacity / ssize; if (c * ssize < initialCapacity) ++c; // 默认 MIN_SEGMENT_TABLE_CAPACITY 是 2，这个值也是有讲究的，因为这样的话，对于具体的槽上， // 插入一个元素不至于扩容，插入第二个的时候才会扩容 int cap = MIN_SEGMENT_TABLE_CAPACITY; while (cap < c) cap <<= 1; // 创建 Segment 数组， // 并创建数组的第一个元素 segment[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]; // 往数组写入 segment[0] UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0] this.segments = ss; }
初始化完成，我们得到了一个 Segment 数组。
我们就当是用 new ConcurrentHashMap() 无参构造函数进行初始化的，那么初始化完成后：
- Segment 数组长度为 16，不可以扩容
- Segment[i] 的默认大小为 2，负载因子是 0.75，得出初始阈值为 1.5，也就是以后插入第一个元素不会触发扩容，插入第二个会进行第一次扩容
- 这里初始化了 segment[0]，其他位置还是 null，至于为什么要初始化 segment[0]，后面的代码会介绍
- 当前 segmentShift 的值为 32 – 4 = 28，segmentMask 为 16 – 1 = 15，姑且把它们简单翻译为移位数和掩码，这两个值马上就会用到
put 过程分析
我们先看 put 的主流程，对于其中的一些关键细节操作，后面会进行详细介绍。
public V put(K key, V value) { Segment<K,V> s; if (value == null) throw new NullPointerException(); // 1. 计算 key 的 hash 值 int hash = hash(key); // 2. 根据 hash 值找到 Segment 数组中的位置 j // hash 是 32 位，无符号右移 segmentShift(28) 位，剩下低 4 位， // 然后和 segmentMask(15) 做一次与操作，也就是说 j 是 hash 值的最后 4 位，也就是槽的数组下标 int j = (hash >>> segmentShift) & segmentMask; // 刚刚说了，初始化的时候初始化了 segment[0]，但是其他位置还是 null， // ensureSegment(j) 对 segment[j] 进行初始化 if ((s = (Segment<K,V>)UNSAFE.getObject // nonvolatile; recheck (segments, (j << SSHIFT) + SBASE)) == null) // in ensureSegment s = ensureSegment(j); // 3. 插入新值到 槽 s 中 return s.put(key, hash, value, false); }
第一层皮很简单，根据 hash 值很快就能找到相应的 Segment，之后就是 Segment 内部的 put 操作了。
Segment 内部是由 数组+链表 组成的。
final V put(K key, int hash, V value, boolean onlyIfAbsent) { // 在往该 segment 写入前，需要先获取该 segment 的独占锁 // 先看主流程，后面还会具体介绍这部分内容 HashEntry<K,V> node = tryLock() ? null : scanAndLockForPut(key, hash, value); V oldValue; try { // 这个是 segment 内部的数组 HashEntry<K,V>[] tab = table; // 再利用 hash 值，求应该放置的数组下标 int index = (tab.length - 1) & hash; // first 是数组该位置处的链表的表头 HashEntry<K,V> first = entryAt(tab, index); // 下面这串 for 循环虽然很长，不过也很好理解，想想该位置没有任何元素和已经存在一个链表这两种情况 for (HashEntry<K,V> e = first;;) { if (e != null) { K k; if ((k = e.key) == key || (e.hash == hash && key.equals(k))) { oldValue = e.value; if (!onlyIfAbsent) { // 覆盖旧值 e.value = value; ++modCount; } break; } // 继续顺着链表走 e = e.next; } else { // node 到底是不是 null，这个要看获取锁的过程，不过和这里都没有关系。 // 如果不为 null，那就直接将它设置为链表表头；如果是null，初始化并设置为链表表头。 if (node != null) node.setNext(first); else node = new HashEntry<K,V>(hash, key, value, first); int c = count + 1; // 如果超过了该 segment 的阈值，这个 segment 需要扩容 if (c > threshold && tab.length < MAXIMUM_CAPACITY) rehash(node); // 扩容后面也会具体分析 else // 没有达到阈值，将 node 放到数组 tab 的 index 位置， // 其实就是将新的节点设置成原链表的表头 setEntryAt(tab, index, node); ++modCount; count = c; oldValue = null; break; } } } finally { // 解锁 unlock(); } return oldValue; }
整体流程还是比较简单的，由于有独占锁的保护，所以 segment 内部的操作并不复杂。至于这里面的并发问题，我们稍后再进行介绍。
到这里 put 操作就结束了，接下来，我们说一说其中几步关键的操作。
初始化槽: ensureSegment
ConcurrentHashMap 初始化的时候会初始化第一个槽 segment[0]，对于其他槽来说，在插入第一个值的时候进行初始化。
这里需要考虑并发，因为很可能会有多个线程同时进来初始化同一个槽 segment[k]，不过只要有一个成功了就可以。
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[0] 了， // 使用当前 segment[0] 处的数组长度和负载因子来初始化 segment[k] // 为什么要用“当前”，因为 segment[0] 可能早就扩容过了 Segment<K,V> proto = ss[0]; int cap = proto.table.length; float lf = proto.loadFactor; int threshold = (int)(cap * lf); // 初始化 segment[k] 内部的数组 HashEntry<K,V>[] tab = (HashEntry<K,V>[])new HashEntry[cap]; if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) { // 再次检查一遍该槽是否被其他线程初始化了。 Segment<K,V> s = new Segment<K,V>(lf, threshold, tab); // 使用 while 循环，内部用 CAS，当前线程成功设值或其他线程成功设值后，退出 while ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) { if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s)) break; } } } return seg; }
总的来说，ensureSegment(int k) 比较简单，对于并发操作使用 CAS 进行控制。
我没搞懂这里为什么要搞一个 while 循环，CAS 失败不就代表有其他线程成功了吗，为什么要再进行判断？ 感谢评论区的李子木，如果当前线程 CAS 失败，这里的 while 循环是为了将 seg 赋值返回。
获取写入锁: scanAndLockForPut
前面我们看到，在往某个 segment 中 put 的时候，首先会调用 node = tryLock() ? null : scanAndLockForPut(key, hash, value)，也就是说先进行一次 tryLock() 快速获取该 segment 的独占锁，如果失败，那么进入到 scanAndLockForPut 这个方法来获取锁。
下面我们来具体分析这个方法中是怎么控制加锁的。
private HashEntry<K,V> scanAndLockForPut(K key, int hash, V value) { HashEntry<K,V> first = entryForHash(this, hash); HashEntry<K,V> e = first; HashEntry<K,V> node = null; int retries = -1; // negative while locating node // 循环获取锁 while (!tryLock()) { HashEntry<K,V> f; // to recheck first below if (retries < 0) { if (e == null) { if (node == null) // speculatively create node // 进到这里说明数组该位置的链表是空的，没有任何元素 // 当然，进到这里的另一个原因是 tryLock() 失败，所以该槽存在并发，不一定是该位置 node = new HashEntry<K,V>(hash, key, value, null); retries = 0; } else if (key.equals(e.key)) retries = 0; else // 顺着链表往下走 e = e.next; } // 重试次数如果超过 MAX_SCAN_RETRIES（单核1多核64），那么不抢了，进入到阻塞队列等待锁 // lock() 是阻塞方法，直到获取锁后返回 else if (++retries > MAX_SCAN_RETRIES) { lock(); break; } else if ((retries & 1) == 0 && // 这个时候是有大问题了，那就是有新的元素进到了链表，成为了新的表头 // 所以这边的策略是，相当于重新走一遍这个 scanAndLockForPut 方法 (f = entryForHash(this, hash)) != first) { e = first = f; // re-traverse if entry changed retries = -1; } } return node; }
这个方法有两个出口，一个是 tryLock() 成功了，循环终止，另一个就是重试次数超过了 MAXSCANRETRIES，进到 lock() 方法，此方法会阻塞等待，直到成功拿到独占锁。
这个方法就是看似复杂，但是其实就是做了一件事，那就是获取该 segment 的独占锁，如果需要的话顺便实例化了一下 node。
扩容: rehash
重复一下，segment 数组不能扩容，扩容是 segment 数组某个位置内部的数组 HashEntry[] 进行扩容，扩容后，容量为原来的 2 倍。
首先，我们要回顾一下触发扩容的地方，put 的时候，如果判断该值的插入会导致该 segment 的元素个数超过阈值，那么先进行扩容，再插值，读者这个时候可以回去 put 方法看一眼。
该方法不需要考虑并发，因为到这里的时候，是持有该 segment 的独占锁的。
// 方法参数上的 node 是这次扩容后，需要添加到新的数组中的数据。 private void rehash(HashEntry<K,V> node) { HashEntry<K,V>[] oldTable = table; int oldCapacity = oldTable.length; // 2 倍 int newCapacity = oldCapacity << 1; threshold = (int)(newCapacity * loadFactor); // 创建新数组 HashEntry<K,V>[] newTable = (HashEntry<K,V>[]) new HashEntry[newCapacity]; // 新的掩码，如从 16 扩容到 32，那么 sizeMask 为 31，对应二进制 ‘000...00011111’ int sizeMask = newCapacity - 1; // 遍历原数组，老套路，将原数组位置 i 处的链表拆分到 新数组位置 i 和 i+oldCap 两个位置 for (int i = 0; i < oldCapacity ; i++) { // e 是链表的第一个元素 HashEntry<K,V> e = oldTable[i]; if (e != null) { HashEntry<K,V> next = e.next; // 计算应该放置在新数组中的位置， // 假设原数组长度为 16，e 在 oldTable[3] 处，那么 idx 只可能是 3 或者是 3 + 16 = 19 int idx = e.hash & sizeMask; if (next == null) // 该位置处只有一个元素，那比较好办 newTable[idx] = e; else { // Reuse consecutive sequence at same slot // e 是链表表头 HashEntry<K,V> lastRun = e; // idx 是当前链表的头结点 e 的新位置 int lastIdx = idx; // 下面这个 for 循环会找到一个 lastRun 节点，这个节点之后的所有元素是将要放到一起的 for (HashEntry<K,V> last = next; last != null; last = last.next) { int k = last.hash & sizeMask; if (k != lastIdx) { lastIdx = k; lastRun = last; } } // 将 lastRun 及其之后的所有节点组成的这个链表放到 lastIdx 这个位置 newTable[lastIdx] = lastRun; // 下面的操作是处理 lastRun 之前的节点， // 这些节点可能分配在另一个链表中，也可能分配到上面的那个链表中 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); } } } } // 将新来的 node 放到新数组中刚刚的 两个链表之一 的 头部 int nodeIndex = node.hash & sizeMask; // add the new node node.setNext(newTable[nodeIndex]); newTable[nodeIndex] = node; table = newTable; }
这里的扩容比之前的 HashMap 要复杂一些，代码难懂一点。上面有两个挨着的 for 循环，第一个 for 有什么用呢？
仔细一看发现，如果没有第一个 for 循环，也是可以工作的，但是，这个 for 循环下来，如果 lastRun 的后面还有比较多的节点，那么这次就是值得的。因为我们只需要克隆 lastRun 前面的节点，后面的一串节点跟着 lastRun 走就是了，不需要做任何操作。
我觉得 Doug Lea 的这个想法也是挺有意思的，不过比较坏的情况就是每次 lastRun 都是链表的最后一个元素或者很靠后的元素，那么这次遍历就有点浪费了。不过 Doug Lea 也说了，根据统计，如果使用默认的阈值，大约只有 1/6 的节点需要克隆。
get 过程分析
相对于 put 来说，get 真的不要太简单。
- 计算 hash 值，找到 segment 数组中的具体位置，或我们前面用的“槽”
- 槽中也是一个数组，根据 hash 找到数组中具体的位置
- 到这里是链表了，顺着链表进行查找即可
public V get(Object key) { Segment<K,V> s; // manually integrate access methods to reduce overhead HashEntry<K,V>[] tab; // 1. hash 值 int h = hash(key); long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE; // 2. 根据 hash 找到对应的 segment if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null && (tab = s.table) != null) { // 3. 找到segment 内部数组相应位置的链表，遍历 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操作
put、remove和get操作只需要关心一个Segment，而size操作需要遍历所有的Segment才能算出整个Map的大小。一个简单的方案是，先锁住所有Sgment，计算完后再解锁。但这样做，在做size操作时，不仅无法对Map进行写操作，同时也无法进行读操作，不利于对Map的并行操作。
为更好支持并发操作，ConcurrentHashMap会在不上锁的前提逐个Segment计算3次size，如果某相邻两次计算获取的所有Segment的更新次数（每个Segment都与HashMap一样通过modCount跟踪自己的修改次数，Segment每修改一次其modCount加一）相等，说明这两次计算过程中无更新操作，则这两次计算出的总size相等，可直接作为最终结果返回。如果这三次计算过程中Map有更新，则对所有Segment加锁重新计算Size。该计算方法代码如下
public int size() { 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 (;;) { if (retries++ == RETRIES_BEFORE_LOCK) { 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; } } 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; }
不同之处
JDK1.7中
ConcurrentHashMap与HashMap相比，有以下不同点
- ConcurrentHashMap线程安全，而HashMap非线程安全
- HashMap允许Key和Value为null，而ConcurrentHashMap不允许
- HashMap不允许通过Iterator遍历的同时通过HashMap修改，而ConcurrentHashMap允许该行为，并且该更新对后续的遍历可见
并发问题分析
现在我们已经说完了 put 过程和 get 过程，我们可以看到 get 过程中是没有加锁的，那自然我们就需要去考虑并发问题。
添加节点的操作 put 和删除节点的操作 remove 都是要加 segment 上的独占锁的，所以它们之间自然不会有问题，我们需要考虑的问题就是 get 的时候在同一个 segment 中发生了 put 或 remove 操作。
- put 操作的线程安全性。
- 初始化槽，这个我们之前就说过了，使用了 CAS 来初始化 Segment 中的数组。
- 添加节点到链表的操作是插入到表头的，所以，如果这个时候 get 操作在链表遍历的过程已经到了中间，是不会影响的。当然，另一个并发问题就是 get 操作在 put 之后，需要保证刚刚插入表头的节点被读取，这个依赖于 setEntryAt 方法中使用的 UNSAFE.putOrderedObject。
- 扩容。扩容是新创建了数组，然后进行迁移数据，最后面将 newTable 设置给属性 table。所以，如果 get 操作此时也在进行，那么也没关系，如果 get 先行，那么就是在旧的 table 上做查询操作；而 put 先行，那么 put 操作的可见性保证就是 table 使用了 volatile 关键字。
- remove 操作的线程安全性。 remove 操作我们没有分析源码，所以这里说的读者感兴趣的话还是需要到源码中去求实一下的。 get 操作需要遍历链表，但是 remove 操作会"破坏"链表。 如果 remove 破坏的节点 get 操作已经过去了，那么这里不存在任何问题。 如果 remove 先破坏了一个节点，分两种情况考虑。 1、如果此节点是头结点，那么需要将头结点的 next 设置为数组该位置的元素，table 虽然使用了 volatile 修饰，但是 volatile 并不能提供数组内部操作的可见性保证，所以源码中使用了 UNSAFE 来操作数组，请看方法 setEntryAt。2、如果要删除的节点不是头结点，它会将要删除节点的后继节点接到前驱节点中，这里的并发保证就是 next 属性是 volatile 的。
Java8 HashMap
Java8 对 HashMap 进行了一些修改，最大的不同就是利用了红黑树，所以其由 数组+链表+红黑树 组成。
根据 Java7 HashMap 的介绍，我们知道，查找的时候，根据 hash 值我们能够快速定位到数组的具体下标，但是之后的话，需要顺着链表一个个比较下去才能找到我们需要的，时间复杂度取决于链表的长度，为 O(n)。
为了降低这部分的开销，在 Java8 中，当链表中的元素达到了 8 个时，会将链表转换为红黑树，在这些位置进行查找的时候可以降低时间复杂度为 O(logN)。
来一张图简单示意一下吧：
注意，上图是示意图，主要是描述结构，不会达到这个状态的，因为这么多数据的时候早就扩容了。
下面，我们还是用代码来介绍吧，个人感觉，Java8 的源码可读性要差一些，不过精简一些。
Java7 中使用 Entry 来代表每个 HashMap 中的数据节点，Java8 中使用 Node，基本没有区别，都是 key，value，hash 和 next 这四个属性，不过，Node 只能用于链表的情况，红黑树的情况需要使用 TreeNode。
我们根据数组元素中，第一个节点数据类型是 Node 还是 TreeNode 来判断该位置下是链表还是红黑树的。
put 过程分析
public V put(K key, V value) { return putVal(hash(key), key, value, false, true); } // 第三个参数 onlyIfAbsent 如果是 true，那么只有在不存在该 key 时才会进行 put 操作 // 第四个参数 evict 我们这里不关心 final V putVal(int hash, K key, V value, boolean onlyIfAbsent, boolean evict) { Node<K,V>[] tab; Node<K,V> p; int n, i; // 第一次 put 值的时候，会触发下面的 resize()，类似 java7 的第一次 put 也要初始化数组长度 // 第一次 resize 和后续的扩容有些不一样，因为这次是数组从 null 初始化到默认的 16 或自定义的初始容量 if ((tab = table) == null || (n = tab.length) == 0) n = (tab = resize()).length; // 找到具体的数组下标，如果此位置没有值，那么直接初始化一下 Node 并放置在这个位置就可以了 if ((p = tab[i = (n - 1) & hash]) == null) tab[i] = newNode(hash, key, value, null); else {// 数组该位置有数据 Node<K,V> e; K k; // 首先，判断该位置的第一个数据和我们要插入的数据，key 是不是"相等"，如果是，取出这个节点 if (p.hash == hash && ((k = p.key) == key || (key != null && key.equals(k)))) e = p; // 如果该节点是代表红黑树的节点，调用红黑树的插值方法，本文不展开说红黑树 else if (p instanceof TreeNode) e = ((TreeNode<K,V>)p).putTreeVal(this, tab, hash, key, value); else { // 到这里，说明数组该位置上是一个链表 for (int binCount = 0; ; ++binCount) { // 插入到链表的最后面(Java7 是插入到链表的最前面) if ((e = p.next) == null) { p.next = newNode(hash, key, value, null); // TREEIFY_THRESHOLD 为 8，所以，如果新插入的值是链表中的第 8 个 // 会触发下面的 treeifyBin，也就是将链表转换为红黑树 if (binCount >= TREEIFY_THRESHOLD - 1) // -1 for 1st treeifyBin(tab, hash); break; } // 如果在该链表中找到了"相等"的 key(== 或 equals) if (e.hash == hash && ((k = e.key) == key || (key != null && key.equals(k)))) // 此时 break，那么 e 为链表中[与要插入的新值的 key "相等"]的 node break; p = e; } } // e!=null 说明存在旧值的key与要插入的key"相等" // 对于我们分析的put操作，下面这个 if 其实就是进行 "值覆盖"，然后返回旧值 if (e != null) { V oldValue = e.value; if (!onlyIfAbsent || oldValue == null) e.value = value; afterNodeAccess(e); return oldValue; } } ++modCount; // 如果 HashMap 由于新插入这个值导致 size 已经超过了阈值，需要进行扩容 if (++size > threshold) resize(); afterNodeInsertion(evict); return null; }
和 Java7 稍微有点不一样的地方就是，Java7 是先扩容后插入新值的，Java8 先插值再扩容，不过这个不重要。
数组扩容
resize() 方法用于初始化数组或数组扩容，每次扩容后，容量为原来的 2 倍，并进行数据迁移。
final Node<K,V>[] resize() { Node<K,V>[] oldTab = table; int oldCap = (oldTab == null) ? 0 : oldTab.length; int oldThr = threshold; int newCap, newThr = 0; if (oldCap > 0) { // 对应数组扩容 if (oldCap >= MAXIMUM_CAPACITY) { threshold = Integer.MAX_VALUE; return oldTab; } // 将数组大小扩大一倍 else if ((newCap = oldCap << 1) < MAXIMUM_CAPACITY && oldCap >= DEFAULT_INITIAL_CAPACITY) // 将阈值扩大一倍 newThr = oldThr << 1; // double threshold } else if (oldThr > 0) // 对应使用 new HashMap(int initialCapacity) 初始化后，第一次 put 的时候 newCap = oldThr; else {// 对应使用 new HashMap() 初始化后，第一次 put 的时候 newCap = DEFAULT_INITIAL_CAPACITY; newThr = (int)(DEFAULT_LOAD_FACTOR * DEFAULT_INITIAL_CAPACITY); } if (newThr == 0) { float ft = (float)newCap * loadFactor; newThr = (newCap < MAXIMUM_CAPACITY && ft < (float)MAXIMUM_CAPACITY ? (int)ft : Integer.MAX_VALUE); } threshold = newThr; // 用新的数组大小初始化新的数组 Node<K,V>[] newTab = (Node<K,V>[])new Node[newCap]; table = newTab; // 如果是初始化数组，到这里就结束了，返回 newTab 即可 if (oldTab != null) { // 开始遍历原数组，进行数据迁移。 for (int j = 0; j < oldCap; ++j) { Node<K,V> e; if ((e = oldTab[j]) != null) { oldTab[j] = null; // 如果该数组位置上只有单个元素，那就简单了，简单迁移这个元素就可以了 if (e.next == null) newTab[e.hash & (newCap - 1)] = e; // 如果是红黑树，具体我们就不展开了 else if (e instanceof TreeNode) ((TreeNode<K,V>)e).split(this, newTab, j, oldCap); else { // 这块是处理链表的情况， // 需要将此链表拆成两个链表，放到新的数组中，并且保留原来的先后顺序 // loHead、loTail 对应一条链表，hiHead、hiTail 对应另一条链表，代码还是比较简单的 Node<K,V> loHead = null, loTail = null; Node<K,V> hiHead = null, hiTail = null; Node<K,V> next; do { next = e.next; if ((e.hash & oldCap) == 0) { if (loTail == null) loHead = e; else loTail.next = e; loTail = e; } else { if (hiTail == null) hiHead = e; else hiTail.next = e; hiTail = e; } } while ((e = next) != null); if (loTail != null) { loTail.next = null; // 第一条链表 newTab[j] = loHead; } if (hiTail != null) { hiTail.next = null; // 第二条链表的新的位置是 j + oldCap，这个很好理解 newTab[j + oldCap] = hiHead; } } } } } return newTab; }
get 过程分析
相对于 put 来说，get 真的太简单了。
- 计算 key 的 hash 值，根据 hash 值找到对应数组下标: hash & (length-1)
- 判断数组该位置处的元素是否刚好就是我们要找的，如果不是，走第三步
- 判断该元素类型是否是 TreeNode，如果是，用红黑树的方法取数据，如果不是，走第四步
- 遍历链表，直到找到相等(==或equals)的 key
public V get(Object key) { Node<K,V> e; return (e = getNode(hash(key), key)) == null ? null : e.value; }
final Node<K,V> getNode(int hash, Object key) { Node<K,V>[] tab; Node<K,V> first, e; int n; K k; if ((tab = table) != null && (n = tab.length) > 0 && (first = tab[(n - 1) & hash]) != null) { // 判断第一个节点是不是就是需要的 if (first.hash == hash && // always check first node ((k = first.key) == key || (key != null && key.equals(k)))) return first; if ((e = first.next) != null) { // 判断是否是红黑树 if (first instanceof TreeNode) return ((TreeNode<K,V>)first).getTreeNode(hash, key); // 链表遍历 do { if (e.hash == hash && ((k = e.key) == key || (key != null && key.equals(k)))) return e; } while ((e = e.next) != null); } } return null; }
