原文： https:///post/hashma
排版：java技术栈
网上关于 HashMap 和 ConcurrentHashMap 的文章确实不少，不过缺斤少两的文章比较多，所以才想自己也写一篇，把细节说清楚说透，尤其像 Java8 中的 ConcurrentHashMap，大部分文章都说不清楚。
终归是希望能降低大家学习的成本，不希望大家到处找各种不是很靠谱的文章，看完一篇又一篇，可是还是模模糊糊。
阅读建议：
Java7 HashMap -> Java7 ConcurrentHashMap -> Java8 HashMap -> Java8 ConcurrentHashMap
阅读前提： Java8 用到了红黑树，不过本文不会进行展开，感兴趣的读者请自行查找相关资料。
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 ); }
扩容就是用一个新的大数组替换原来的小数组，并将原来数组中的值迁移到新的数组中。
由于是双倍扩容，迁移过程中，会将原来 table[i] 中的链表的所有节点，分拆到新的数组的 newTable[i] 和 newTable[i + oldLength] 位置上。 如原来数组长度是 16，那么扩容后，原来 table[0] 处的链表中的所有元素会被分配到新数组中 newTable[0] 和 newTable[16] 这两个位置。 代码比较简单，这里就不展开了。
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 是线程安全的，也就实现了全局的线程安全。 推荐阅读： 史上最全 Java 多线程面试题及答案 。
concurrencyLevel： 默认是 16，也就是说 ConcurrentHashMap 有 16 个 Segments，所以理论上，这个时候，最多可以同时支持 16 个线程并发写，只要它们的操作分别分布在不同的 Segment 上。 这个值可以在初始化的时候设置为其他值，但是一旦初始化以后，它是不可以扩容的。
再具体到每个 Segment 内部，其实每个 Segment 很像之前介绍的 HashMap，不过它要保证线程安全，所以处理起来要麻烦些。 推荐阅读： 几种线程安全的Map解析 。
初始化 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 进行控制。
如果当前线程 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() 成功了，循环终止，另一个就是重试次数超过了 MAX_SCAN_RETRIES，进到 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 真的不要太简单。
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 ; } 并发问题分析 现在我们已经说完了 put 过程和 get 过程，我们可以看到 get 过程中是没有加锁的，那自然我们就需要去考虑并发问题。
