JDK1.7、JDK1.8的HashMap、ConcurrentHashMap笔记

梦里梦外; 2023-01-06 15:55 86阅读 0赞

### 目录 ###

*  1 简介
 *  2. HashMap
 *   *  2.1 JDK1.7中的实现
     *   *  2.1.1 设计思想
         *  2.1.2 Entry的定义
         *  2.1.3 Entry数组
         *   *  2.1.3.1 threshold和loadFactor
             *  2.1.3.2 扩容
             *   *  2.1.3.2.1 扩容条件
                 *  2.1.3.2.2 扩容过程
                 *  2.1.3.2.3 扩容时的rehash
                 *  2.1.3.2.4 扩容存在的问题
         *  2.1.4 计算数组下标
         *   *  2.1.4.1 哈希函数
             *   *  2.1.4.1.1 为什么要对hashSeed为0时的String类型的对象做特殊处理？
                 *  2.1.4.1.2 为什么要采用位运算？
                 *  2.1.4.1.3 把上述的位移运算移动的位数稍微改一下会怎么样？
             *  2.1.4.2 根据哈希值确定下标
         *  2.1.5 put()函数
         *  2.1.6 get()函数
         *  2.1.7 fail-fast机制
         *  2.1.8 keySey()、entrySet()和values()
         *  2.1.9 总结
     *  2.2 JDK1.8中的实现
     *   *  2.2.1 设计思想
         *  2.2.2 Node节点的定义
         *  2.2.3 哈希函数
         *  2.2.4 扩容过程
         *   *  2.2.4.1 为什么原table数组中下标为i下的节点只会转移到新数组下标i和(i + n)处？
             *  2.2.4.2 头插法效率比尾插法要更高，JDK1.8为什么要将头插法变为尾插法？
         *  2.2.5 节点树化和去树化的条件
         *   *  2.2.5.1 为什么链表长度为8时发生树化？
             *  2.2.5.2 为什么链表长度为6时发生去树化？
             *  2.2.5.3 以什么作为红黑树节点的比较条件？
         *  2.2.6 总结
 *  3. ConcurrentHashMap
 *   *  3.1 JDK1.7中的实现
     *   *  3.1.1 设计思想
         *  3.1.2 HashEntry的定义
         *  3.1.2 Segment的定义
         *   *  3.1.2.1 scanAndLockForPut()函数和scanAndLock()函数
             *  3.1.2.2 rehash()函数
             *   *  3.1.2.2.1 为什么需要克隆HashEntry节点？为什么仅需要克隆部分HashEntry节点？
         *  3.1.3 构造函数
         *   *  3.1.3.1 为什么在构造函数中仅初始化下标为0的Segment？
             *  3.1.3.2 为什么在构造函数中调用UNSAFE.putOrderedObject()方法给segments数组的0号下标初始化？
         *  3.1.4 lazySet
         *  3.1.5 聚合函数的实现
         *  3.1.6 总结
     *  3.2 JDK1.8中的实现
     *   *  3.2.1 设计思想
         *  3.2.2 节点定义
         *  3.2.3 sizeCtl的含义
         *  3.2.4 哈希函数
         *  3.2.5 put()函数
         *   *  3.2.5.1 initTable()函数
             *  3.2.5.2 helpTransfer()函数
             *   *  3.2.5.2.1 一个小错误
                 *   *  3.2.5.2.1.1 条件一：(sc >>> RESIZE\_STAMP\_SHIFT) != rs
                     *  3.2.5.2.1.2 条件二：sc == rs + 1
                     *  3.2.5.2.1.3 条件三：sc == rs + MAX\_RESIZERS
                     *  3.2.5.2.1.4 条件四：transferIndex <= 0
                 *  3.2.5.2.2 transfer()函数
             *  3.2.5.3 addCount()函数
         *  3.2.6 get()函数
         *   *  3.2.6.1 TreeBin中的find()函数
             *  3.2.6.2 ForwardingNode中的find()函数
             *  3.2.6.3 ReservationNode中的find()函数
             *  3.2.6.4 Node中的find()函数
         *  3.2.7 迭代器的实现
         *   *  3.2.7.1 设计思想
             *  3.2.7.2 类结构
             *  3.2.7.3 缺陷
         *  3.2.8 总结
 *  4. 写在最后

# 1 简介 #

HashMap是一种存储键值对的数据结构，设计目的是能够以O(1)的时间复杂度存储键值对和根据键获取值。

在HashMap出现之前，Java程序员只能使用Hashtable。Hashtable是一个线程安全的HashMap，put()和get()方法均使用了synchronized关键字对整个Hashtable加锁。在不需要线程安全保证的情况下，Hashtable的性能不如HashMap；但在需要线程安全保证的情况下，Hashtable的性能不如ConcurrentHashMap。因此，无论是否需要保证线程安全性，Hashtable都已经不再被建议使用了。

从设计的角度来说，Hashtable是JDK设计者的一个失败的设计，强制用synchronized加全局锁的形式保证了线程安全性。而现在Hashtable被HashMap和ConcurrentHashMap全面代替，HashMap完全不加锁，不保证线程安全性，ConcurrentHashMap采用分段锁的思想实现，保证了线程安全性。根据开发者的实际需要，可以采用HashMap和ConcurrentHashMap，而不是将线程安全性强制绑定在Hashtable中。

# 2. HashMap #

在JDK1.7及其之前的版本中，HashMap的底层数据结构是数组+链表。在JDK1.8及其之后的版本中，HashMap的底层数据结构是数组+链表+红黑树。

## 2.1 JDK1.7中的实现 ##

### 2.1.1 设计思想 ###

*  一个健值对被封装成一个Entry对象。
 *  使用一个Entry数组来存储所有健值对。
 *  根据key来计算哈希值确定该键值对在Entry数组中的下标。
 *  利用链表法来解决哈希冲突。
 *  当健值对数目过多时，Entry数组会扩容。

### 2.1.2 Entry的定义 ###

final K key;
    V value;
    Entry<K,V> next;
    int hash;

发生哈希冲突时，在Entry数组的同一个下标处会放置多个Entry，这多个Entry是通过next属性链接成一个链表结构的。hash存储的是key对应的哈希值，因为哈希函数的计算过程一般比较耗时，所以将其存储在Entry中避免多次重复计算。

### 2.1.3 Entry数组 ###

static final Entry<?,?>[] EMPTY_TABLE = {
        };
    transient Entry<K,V>[] table = (Entry<K,V>[]) EMPTY_TABLE;

用构造函数新建一个空的HashMap时，其table数组的大小是0，在调用put()方法往该HashMap中存储值时，table数组才会被初始化。

#### 2.1.3.1 threshold和loadFactor ####

当table数组是EMPTY\_TABLE时，threshold用来计算第一次调用put()方法时table数组需要扩容的大小。

在其他情况下，threshold是table数组的扩容阈值，其值等于capacity乘以loadFactor，其中capacity指的是table数组的大小，loadFactor指的是HashMap的负载率。

threshold和loadFactor可以通过HashMap的构造函数显示指定，如果不指定，则采用默认值，threshold的默认值是16，loadFactor的默认值是0.75。

loadFactor默认值选取为0.75是一种时间复杂度和空间复杂度的平衡。过高的loadFactor会导致table数组中某个下标下的链表长度过长，在get()操作的时候时间复杂度过高；过低的loadFactor会导致table数组频繁扩容，浪费更多的空间。

#### 2.1.3.2 扩容 ####

##### 2.1.3.2.1 扩容条件 #####

*  插入新节点前的键值对数目size 大于等于扩容阈值 threshold。
 *  新节点存放的数组下标不为 null。

##### 2.1.3.2.2 扩容过程 #####

*  新建一个数组，大小是原table数组的2倍。
 *  将原table数组中的Entry都转移至新的数组中。

##### 2.1.3.2.3 扩容时的rehash #####

由于哈希函数的计算比较耗时，在扩容过程中，一般不会重新计算Entry的hash属性，但某些情况下也是可能重新计算Entry的hash属性的。具体是否进行rehash由initHashSeedAsNeeded(int capacity)方法的返回值决定，其中传入的参数capacity是扩容时新数组的大小。

final boolean initHashSeedAsNeeded(int capacity) {
        
        boolean currentAltHashing = hashSeed != 0;
        boolean useAltHashing = sun.misc.VM.isBooted() &&
                (capacity >= Holder.ALTERNATIVE_HASHING_THRESHOLD);
        boolean switching = currentAltHashing ^ useAltHashing;
        if (switching) {
        
            hashSeed = useAltHashing
                ? sun.misc.Hashing.randomHashSeed(this)
                : 0;
        }
        return switching;
    }

默认情况下，hashSeed为0，即currentAltHashing为false。

我们可以通过设置系统变量jdk.map.althashing.threshold来设置Holder.ALTERNATIVE\_HASHING\_THRESHOLD的值。

*  当我们不设置jdk.map.althashing.threshold或者设置jdk.map.althashing.threshold为-1时，Holder.ALTERNATIVE\_HASHING\_THRESHOLD的值为Integer.MAX\_VALUE。
 *  当我们设置jdk.map.althashing.threshold为其余负数时，HashMap会抛出IllegalArgumentException异常。
 *  当我们设置jdk.map.althashing.threshold为正数时，其值就是Holder.ALTERNATIVE\_HASHING\_THRESHOLD的值。

也就是说默认情况下，useAltHashing始终都是false。

因此，在默认情况下switching是false，也就是不开启rehash过程。

如果我们设置jdk.map.althashing.threshold为一个不大的正数，当扩容时新数组的大小大于等于该值时，useAltHashing会变为true，switching也会变为true。此时就会改变hashSeed的值，从而进行rehash过程。

##### 2.1.3.2.4 扩容存在的问题 #####

因为JDK1.7版本的节点采用头插法，所以在扩容时链表节点在原table中的顺序和新数组中的顺序是相反的。

也正是因为采用头插法，所以在多线程同时扩容时可能出现循环链表，使得扩容过程陷入死循环。

循环链表如何出现？这里列举一种情况：

void transfer(Entry[] newTable, boolean rehash) {
        
        int newCapacity = newTable.length;
        for (Entry<K,V> e : table) {
        
            while(null != e) {
        
                Entry<K,V> next = e.next;	// 语句1
                if (rehash) {
        
                    e.hash = null == e.key ? 0 : hash(e.key);
                }
                int i = indexFor(e.hash, newCapacity);
                e.next = newTable[i];
                newTable[i] = e;
                e = next;
            }
        }
    }

当线程1和线程2同时执行到语句1处，线程1继续执行，而线程2在执行完语句1后失去了CPU时间片暂停执行。在线程1执行完整个扩容过程后，线程2获得CPU时间片继续执行，此时线程2的扩容过程就会出现循环链表陷入死循环。

但我们从线程安全的角度来看，这似乎并不是一个问题，HashMap本就设计为一个线程不安全的类，多线程访问时我们需要对整个HashMap加锁，因此循环链表的问题也就不会出现了。

### 2.1.4 计算数组下标 ###

#### 2.1.4.1 哈希函数 ####

final int hash(Object k) {
        
        int h = hashSeed;
        if (0 != h && k instanceof String) {
        
            return sun.misc.Hashing.stringHash32((String) k);
        }
        h ^= k.hashCode();
        h ^= (h >>> 20) ^ (h >>> 12);
        return h ^ (h >>> 7) ^ (h >>> 4);
    }

简单总结一下上述哈希函数的逻辑：

*  如果哈希种子（hashSeed）不为0，且传入的是一个String类型的对象，调用一种特殊的哈希函数来计算哈希值。
 *  如果不满足上述条件，会对哈希种子进行一系列诡异的位运算，包括与传入对象的hashCode异或，位移等等。

计算一个对象的哈希值的计算量着实不小，我们也不知道JDK的设计者为什么要这样计算，但是知道了这个函数是一个哈希函数，我们可以反推测下述几个问题。

##### 2.1.4.1.1 为什么要对hashSeed为0时的String类型的对象做特殊处理？ #####

推测是因为String类型的hashCode计算容易产生哈希冲突，故而调用sun.misc.Hashing.stringHash32()函数来做特殊处理。而hashSeed是一个开关，一般情况下hashSeed为0，这个开关是关上的。我们可以设置环境变量jdk.map.althashing.threshold来打开这个开关。具体如何设置该值，见**2.1.3.2.3 扩容时的rehash**。

##### 2.1.4.1.2 为什么要采用位运算？ #####

推测是哈希函数要求极高的计算效率，位运算更接近计算机底层的数据表示方法，计算效率更高。

##### 2.1.4.1.3 把上述的位移运算移动的位数稍微改一下会怎么样？ #####

如何设计哈希冲突少、计算速度快的哈希函数本来就是一个玄学问题，将上述位移运算的位数稍微改一下影响不大。我推测JDK里取的这几个位移值一定是经过了大量的测试，在哈希冲突少和计算速度快之间取得了良好的平衡，是一种最佳实践。

#### 2.1.4.2 根据哈希值确定下标 ####

假设我们的数组长度是n，那么如何才能使哈希值均匀地分布在数组的每一个位置上，以尽可能地减少哈希冲突呢？取模运算！如何高效地进行取模运算呢？如果我们将数组的长度n规定为2的整数次幂，那么将哈希值与(n - 1)进行位与运算就是对n进行取模运算！这就解释了为什么数组的长度必须是2的整数次幂，而且扩容的时候每次都将数组大小翻倍。

### 2.1.5 put()函数 ###

在**2.1.3 Entry数组**中已经提到，HashMap刚初始化的时候table数组的长度为0，而是在第一次往HashMap中添加值的时候table数组才会有容量，这个过程在函数inflateTable()里：

private void inflateTable(int toSize) {
        
        int capacity = roundUpToPowerOf2(toSize);
        threshold = (int) Math.min(capacity * loadFactor, MAXIMUM_CAPACITY + 1);
        table = new Entry[capacity];
        initHashSeedAsNeeded(capacity);
    }

其中，roundUpToPowerOf2(toSize)的作用是寻找到大于等于toSize的最小值，该值恰好是2的整数次幂。而initHashSeedAsNeeded(capacity)方法已经在**2.1.3.2.3 扩容时的rehash**中分析，是初始化hashSeed用的。

和Hashtable和ConcurrentHashMap不同，HashMap是允许存放null为key和value值的，key为null的Entry一定存放在table数组下标为0的位置，因此在put()函数中会对key为null的情况做特殊处理，调用putForNullKey()方法来添加key为null的Entry：

private V putForNullKey(V value) {
        
        for (Entry<K,V> e = table[0]; e != null; e = e.next) {
        
            if (e.key == null) {
        
                V oldValue = e.value;
                e.value = value;
                e.recordAccess(this);
                return oldValue;
            }
        }
        modCount++;
        addEntry(0, null, value, 0);
        return null;
    }

在这之后，put()方法会根据2.4 计算数组下标所描述的那样，计算key对应的数组下标，将其用头插法插入对应的链表。在插入的过程中可能会发生**2.1.3.2 扩容**中描述的数组扩容操作。

### 2.1.6 get()函数 ###

和put()函数相对应，get()函数也会对key为null的情况做特殊处理，这段逻辑在getForNullKey()方法里：

private V getForNullKey() {
        
        if (size == 0) {
        
            return null;
        }
        for (Entry<K,V> e = table[0]; e != null; e = e.next) {
        
            if (e.key == null)
                return e.value;
        }
        return null;
    }

之后，根据key计算出数组下标，然后遍历对应的链表寻找到正确的value值，比较key是否相同是，先比较hash值是否相同，再用==比较是否是同一个对象，再用equals()方法比较两个对象是否相同。

### 2.1.7 fail-fast机制 ###

当我们在用迭代器遍历HashMap时，如果对HashMap做了结构性修改，如添加一个Entry或者删除一个Entry，将会抛出ConcurrentModificationException异常。

HashMap使用一个成员变量modCount来实现这个机制：modCount初始化为0，每次添加一个Entry或者删除一个Entry都会使modCount加1。当创建迭代器的时候会记录当前的modCount值到迭代器的成员变量expectedModCount中，在迭代过程中一旦发现modCount和expectedModCount不相等，就会抛出ConcurrentModificationException异常。

当然，在迭代的过程中我们可以调用迭代器的remove()方法来删除HashMap中的元素，因为在迭代器的remove()方法中重新读取modCount值到成员变量expectedModCount中，其代码如下：

public void remove() {
        
        if (current == null)
            throw new IllegalStateException();
        if (modCount != expectedModCount)
            throw new ConcurrentModificationException();
        Object k = current.key;
        current = null;
        HashMap.this.removeEntryForKey(k);
        expectedModCount = modCount;	// 重新给expectedModCount赋值，防止抛出ConcurrentModificationException异常
    }

### 2.1.8 keySey()、entrySet()和values() ###

HashMap的keySet()、entrySet()和values()方法返回的不是HashMap中key的一份拷贝、entry的一份拷贝和value的一份拷贝，而是和HashMap指向同一个内存地址的一份引用。因此如果对keySet()、entrySet()和values()的返回值删除元素会影响原有的HashMap。（不支持添加元素操作）

### 2.1.9 总结 ###

将table数组绘制成一张图：

![img][]

JDK1.7的HashMap源码总共1189行，数据结构是数组+链表，为提高运算效率，应用了大量移位运算。同时，在多线程扩容时，因为采用了头插法，可能产生循环链表，造成 transfer() 方法陷入死循环。当然，本文只是枚举了循环链表产生的一种情况，还有其余情况可能会产生循环链表问题。

## 2.2 JDK1.8中的实现 ##

### 2.2.1 设计思想 ###

JDK1.8的HashMap设计思想和JDK1.7的基本相同，但是引入了红黑树这一新的数据结构来提高哈希冲突过于严重时的性能。另外，JDK1.8引入了接口default方法、lambda表达式和Spliterator等新的概念，在HashMap中也添加了相应的方法和新的内部类。

### 2.2.2 Node节点的定义 ###

在JDK1.7中key-value被封装成为Entry，而在JDK1.8中被封装成为Node，这两者除了名字不一样，其余都是相同的。Node节点依然拥有下述四个属性：

final int hash;
    final K key;
    V value;
    Node<K,V> next;

由于JDK1.8新引入了红黑树节点，因此还多了一种Node的子类TreeNode，其中增加了5个属性：

TreeNode<K,V> parent;  // red-black tree links
    TreeNode<K,V> left;
    TreeNode<K,V> right;
    TreeNode<K,V> prev;    // needed to unlink next upon deletion
    boolean red;

其中parent用来标记节点的父亲节点，left和right分别用来标记节点的左孩子和右孩子，而prev则用来标记节点的前一个节点（由于TreeNode继承自Node，因此也有next属性，顺着next属性遍历可以得到一条链表，比如A.next = C，那么有C.prev = A，这就是prev属性的作用），red用来标记节点颜色是否是红色。

关于许多红黑树的操作都封装在TreeNode中，由于红黑树的各种操作过于复杂且过于死板，在这里就不再分析。我们只需要知道红黑树是一种二叉搜索树，在其中寻找值和添加值的时间复杂度都是O(logn)，至于具体实现，不是HashMap的重点。

### 2.2.3 哈希函数 ###

我们可以回到2.4.1 哈希函数回顾一下JDK1.7的哈希函数，可以看到特别复杂，而JDK1.8的哈希函数得到了简化：

static final int hash(Object key) {
        
        int h;
        return (key == null) ? 0 : (h = key.hashCode()) ^ (h >>> 16);
    }

这里用h异或运算(h >>> 16)，主要是为了让h的高位参与哈希函数的运算。在**2.1.4.2 根据哈希值确定下标**中我们知道最后是利用哈希函数得到的值和table数组长度减1做位与运算得到数组下标的，也就是说只有哈希值的低位在起作用，以table数组长度为16举例，此时table数组减1的二进制表示法为00000000 00000000 00000000 00001111，此时只有哈希值的低4位会对数组下标产生影响。为了让高位也能起作用，将h右移16位和h做异或运算得到的结果作为哈希值。

之所以JDK1.8的哈希函数能比JDK1.7的简化许多，是因为JDK1.8引入了红黑树这一数据结构，对于哈希冲突过于严重的情况有了很好的处理，因此可以适当地简化哈希函数来提高性能，毕竟JDK1.7中计算哈希值是一个比较耗性能的操作。

### 2.2.4 扩容过程 ###

JDK1.8的HashMap的整个扩容过程和JDK1.7是相同的：

*  新建一个大小为原table数组2倍的新数组。
 *  将原table数组中每个下标下的节点都转移至 新数组中。

但在具体实现细节上，JDK1.8相比JDK1.7有略微不同：

*  JDK1.7采用的是链表的头插法，新节点插在链表头部，在扩容时会导致链表反转；而JDK1.8采用的是链表的尾插法，新节点插在链表的尾部，在扩容前后链表不会反转。
 *  JDK1.7是从原table数组中一个一个Entry地转移到新数组，且可能发生rehash操作。对于链表节点，JDK1.8中是先将原table数组中某个下标i下的节点进行分类，一部分节点转移至新数组的下标i位置，另一部分节点转移至新数组的下标(i + n)位置，其中n为原table数组的大小，将这两部分节点分别构建成链表后再转移至新数组对应的下标位置i和(i + n)处。而对于红黑树节点，也是和链表节点相同的操作，但需要额外维护一个prev指针，再根据新生成的两个链表中的节点个数是否小于等于UNTREEIFY\_THRESHOLD来判断是否需要将红黑树退化为普通的链表节点Node。另外，JDK1.8在扩容时不会发生rehash。

#### 2.2.4.1 为什么原table数组中下标为i下的节点只会转移到新数组下标i和(i + n)处？ ####

因为扩容操作每次将数组大小变为原数组大小的2倍，从而始终保证了数组大小为2的整数次幂。

假设原数组大小为16，那么计算数组下标时，是和00000000 00000000 00000000 00001111做位与运算。

扩容后，计算数组下标时，是和00000000 00000000 00000000 00011111做位与运算，扩容前后的下标值只有最低的第4位会有差别，即要么依然是下标i，要么是下标(i + 16)。

#### 2.2.4.2 头插法效率比尾插法要更高，JDK1.8为什么要将头插法变为尾插法？ ####

在**2.1.3.2.4 扩容存在的问题**中我们已经分析过，在多线程环境下扩容，JDK1.7可能会出现循环链表从而陷入死循环。而JDK1.8采用尾插法这一机制后就避免了这个问题。

另一方面，JDK1.8引入了红黑树的概念，当某个链表中节点数过多时，会转化成红黑树节点，而红黑树节点的插入效率是O(logn)，这就保证了即使是尾插法，效率也不会比头插法差太多。

### 2.2.5 节点树化和去树化的条件 ###

在插入新节点前，某个数组下标的链表长度大于等于TREEIFY\_THRESHOLD（默认值为8）时，将新节点插入到链表尾部，并对整个链表进行树化操作，所以真正发生树化操作时，链表的长度为9。树化操作定义在函数treeifyBin()中：

final void treeifyBin(Node<K,V>[] tab, int hash) {
        
        int n, index; Node<K,V> e;
        if (tab == null || (n = tab.length) < MIN_TREEIFY_CAPACITY)
            resize();
        else if ((e = tab[index = (n - 1) & hash]) != null) {
        
            TreeNode<K,V> hd = null, tl = null;
            do {
        
                TreeNode<K,V> p = replacementTreeNode(e, null);
                if (tl == null)
                    hd = p;
                else {
        
                    p.prev = tl;
                    tl.next = p;
                }
                tl = p;
            } while ((e = e.next) != null);
            if ((tab[index] = hd) != null)
                hd.treeify(tab);
        }
    }

上述代码的逻辑概括如下：

*  如果tab数组还没有被初始化，或者tab数组的长度小于MIN\_TREEIFY\_CAPACITY（默认值是64），将进行数组的扩容操作而不是将链表节点转化为红黑树节点。
 *  确认了tab数组长度大于等于MIN\_TREEIFY\_CAPACITY后，先将原链表节点全部转化为TreeNode的双向链表，再将双向链表转化为红黑树。

去树化在**2.1.3.2 扩容**过程中已有提及。

#### 2.2.5.1 为什么链表长度为8时发生树化？ ####

在key的hashCode()实现良好的基础上，默认负载率loadFactor为0.75时，某个数组下标下Node数目服从下述泊松分布：

0:    0.60653066
    1:    0.30326533
    2:    0.07581633
    3:    0.01263606
    4:    0.00157952
    5:    0.00015795
    6:    0.00001316
    7:    0.00000094
    8:    0.00000006

可以看到，某个数组下标下Node数目为8的概率仅有0.00000006，极低，因此树化的操作并不会经常发生，因为将Node转化为TreeNode是一个耗费时间和空间的过程。

但是如果某个数组下标下Node数目到达了8，继续往该下标下新增节点时，就会发生树化操作。因为检索链表的时间复杂度是O(n)，而检索红黑树的时间复杂度是O(logn)。

#### 2.2.5.2 为什么链表长度为6时发生去树化？ ####

去树化界限为6，比树化界限8要小，在树化和去树化之间留有一定的阈值主要是为了避免频繁地发生树化和去树化操作，因为这两个操作比较耗时。

#### 2.2.5.3 以什么作为红黑树节点的比较条件？ ####

*  以Node对应的hash值作为比较条件。
 *  如果Node对应的hash值相等，且key实现了Comparable接口，则调用其compareTo()方法作为比较条件。
 *  如果Node对应的hash值相等，且key没有实现Comparable接口，或者compareTo()的比较结果是相等的，则用System.identityHashCode()作为比较条件，每个key对象的System.identityHashCode()是不可能相等的。

### 2.2.6 总结 ###

将table数组绘制成一张图：

![img][img 1]

JDK1.8的HashMap总共2398行，数据结构是数组+链表+红黑树，位运算出神入化。

# 3. ConcurrentHashMap #

ConcurrentHashMap是保证线程安全的HashMap，但不是绝对线程安全，仅仅是相对的线程安全，即对ConcurrentHashMap单次的操作是线程安全的，我们在调用的时候不需要进行额外的保障措施，但是对于一些特定顺序的连续调用，就可能需要在调用端使用额外的同步手段来保证调用的正确性。

为了说明ConcurrentHashMap不是绝对线程安全的，我们来看下面这个例子：

public class HashMapConcurrencyTest1 {
        
    
        public static void main(String[] args) throws InterruptedException {
        
            testConcurrency(new HashMap<>());	// 11209
            testConcurrency(new ConcurrentHashMap<>());	// 11010
        }
    
        private static void testConcurrency(Map<Integer, Integer> map) throws InterruptedException {
        
            map.put(1, 0);
            Runnable runnable = () -> {
        
                for (int i = 0; i < 10000; i++) {
        
                    map.put(1, map.get(1) + 1);	// 语句1
                }
            };
            Thread a = new Thread(runnable, "threadA");
            Thread b = new Thread(runnable, "threadB");
            a.start();
            b.start();
            a.join();
            b.join();
            System.out.println(map.get(1));
        }
    
    }

从上述运行结果我们可以看到，无论是HashMap还是ConcurrentHashMap都不是绝对线程安全的（理论上正确的输出结果应该是20000）。HashMapConcurrencyTest1出问题的根本原因在于，即使是ConcurrentHashMap也无法保证下述语句的线程安全性，因为这涉及到两个独立的操作put()和get()。如果要保证绝对线程安全需要用synchronized()加锁，改进后的代码如下：

public class HashMapConcurrencyTest2 {
        
    
        public static void main(String[] args) throws InterruptedException {
        
            testConcurrency(new HashMap<>());	// 20000
            testConcurrency(new ConcurrentHashMap<>()); // 20000
        }
    
        private static void testConcurrency(Map<Integer, Integer> map) throws InterruptedException {
        
            map.put(1, 0);
            Runnable runnable = () -> {
        
                synchronized (map) {
        
                    for (int i = 0; i < 10000; i++) {
        
                        map.put(1, map.get(1) + 1);	
                    }
                }
            };
            Thread a = new Thread(runnable, "threadA");
            Thread b = new Thread(runnable, "threadB");
            a.start();
            b.start();
            a.join();
            b.join();
            System.out.println(map.get(1));
        }
    
    }

上述例子体现不出HashMap和ConcurrentHashMap在线程安全性上的不同，只能说明两者都不是绝对线程安全的，但不能说明ConcurrentHashMap是相对线程安全的，HashMap不是相对线程安全的。我们再来看一个例子：

public class HashMapConcurrencyTest3 {
        
    
        public static void main(String[] args) throws InterruptedException {
        
            test(new HashMap<>());	// 973
            test(new ConcurrentHashMap<>());	// 1000
        }
    
        private static void test(Map<String, String> map) throws InterruptedException {
        
            CountDownLatch countDownLatch = new CountDownLatch(10);
            Runnable runnable = () -> {
        
                for (int i = 0; i < 100; i++) {
        
                    String s = Thread.currentThread().getName() + i;
                    map.put(s, s);
                }
                countDownLatch.countDown();
            };
            for (int i = 0; i < 10; i++) {
        
                new Thread(runnable, "Thread" + i + ": ").start();
            }
            countDownLatch.await();
            System.out.println(map.size());
      	}
    
    }

在HashMapConcurrencyTest3中，HashMap表现出了线程不安全性，而ConcurrentHashMap则表现出了线程安全性，解释如下：

发生哈希冲突时，2个线程在HashMap的table数组同一个下标位置插入链表节点时，可能出现线程1执行node1.next = node2后，让出时间片，线程2执行node1.next = node3，这样node2节点就丢失了，就表现出了线程不安全性。而ConcurrentHashMap则保证了单次put()操作的线程安全性，是利用分段锁的思想实现的，一次只会有一个线程进入table数组的同一个下标位置进行插入节点的操作，也就实现了线程安全性。

JDK1.7和JDK1.8的ConcurrentHashMap都采用分段锁的思想实现，但是JDK1.8的分段锁的锁粒度更小，且引入了红黑树这一数据结构优化哈希冲突时ConcurrentHashMap的性能。和Hashtable一样，ConcurrentHashMap不允许存放key为null或者value为null的键值对。

## 3.1 JDK1.7中的实现 ##

### 3.1.1 设计思想 ###

核心思想是分段锁，具体阐述如下：

*  将HashMap中的Entry数组用Segment数组来代替，Segment继承自ReentrantLock，是加锁的最小维度。
 *  每一个Segment持有一个HashEntry数组，该HashEntry数组相当于HashMap中的table数组。每个Segment持有的HashEntry数组大小不一定一样大。
 *  将HashMap根据Segment分成若干段，put()和get()函数根据key先定位到Segment数组的下标，再定位到该Segment中HashEntry数组中的下标。
 *  put()函数只允许一个线程对同一个Segment进行操作，这就保证了put()操作的相对线程安全性。

### 3.1.2 HashEntry的定义 ###

HashEntry是ConcurrentHashMap中的一个静态内部类，有4个属性：

final int hash;
    final K key;
    volatile V value;
    volatile HashEntry<K,V> next;

value和next属性被定义成volatile，是为了保证并发情况下的可见性，关于volatile的内存意义，不是本文的重点。

### 3.1.2 Segment的定义 ###

Segment继承自ReentrantLock，拥有以下几个属性：

transient volatile HashEntry<K,V>[] table;
    transient int count;
    transient int modCount;
    transient int threshold;
    final float loadFactor;

我们完全可以把一个Segment当作一个HashMap来对待，其中table数组被定义成volatile，保证了并发情况下的可见性。count标记Segment持有的键值对数目。modCount、threshold和loadFactor和HashMap中的含义相同，modCount表示Segment发生结构性修改的次数，threshold表示Segment中的table数组的扩容阈值，loadFactor表示Segment的负载率。table、count、modCount和threshold都被标记为transient，在默认的序列化实现中这4个参数会被忽略。

#### 3.1.2.1 scanAndLockForPut()函数和scanAndLock()函数 ####

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
                        node = new HashEntry<K,V>(hash, key, value, null);
                    retries = 0;
                }
                else if (key.equals(e.key))
                    retries = 0;
                else
                    e = e.next;
            }
            else if (++retries > MAX_SCAN_RETRIES) {
        
                lock();
                break;
            }
            else if ((retries & 1) == 0 &&
                     (f = entryForHash(this, hash)) != first) {
        
                e = first = f; // re-traverse if entry changed
                retries = -1;
            }
        }
        return node;
    }
    
    private void scanAndLock(Object key, int hash) {
        
        HashEntry<K,V> first = entryForHash(this, hash);
        HashEntry<K,V> e = first;
        int retries = -1;
        while (!tryLock()) {
        
            HashEntry<K,V> f;
            if (retries < 0) {
        
                if (e == null || key.equals(e.key))
                    retries = 0;
                else
                    e = e.next;
            }
            else if (++retries > MAX_SCAN_RETRIES) {
        
                lock();
                break;
            }
            else if ((retries & 1) == 0 &&
                     (f = entryForHash(this, hash)) != first) {
        
                e = first = f;
                retries = -1;
            }
        }
    }

scanAndLockForPut()和scanAndLock()长得差不多，作用也差不多，都是用于获取锁时的线程自旋操作。这个自旋受到最大自旋次数MAX\_SCAN\_RETRIES的限制，如果自旋次数大于MAX\_SCAN\_RETRIES，会直接调用lock()方法阻塞当前线程直到获取锁。在自旋的过程中，每隔1次会检查一下链表的首节点是否发生了变化，如果发生了变化说明有其他线程获取到了锁在修改了链表首节点，需要重新获取链表首节点并重置自旋次数重新进入自旋流程。

区别在于scanAndLockForPut()用于put()函数的自旋操作，有返回值，如果key在table数组中存在，则返回null；如果不存在，返回一个新的HashEntry节点。实际上这个返回值返回什么对put()函数不会有任何的影响，具体可以见put()函数里的逻辑：

final V put(K key, int hash, V value, boolean onlyIfAbsent) {
        
        HashEntry<K,V> node = tryLock() ? null :
            scanAndLockForPut(key, hash, value);
        V oldValue;
        try {
        
            HashEntry<K,V>[] tab = table;
            int index = (tab.length - 1) & hash;
            HashEntry<K,V> first = entryAt(tab, index);
            for (HashEntry<K,V> e = first;;) {
        
                if (e != null) {
        
                    K k;
                    if ((k = e.key) == key ||
                        (e.hash == hash && key.equals(k))) {
        	// 如果scanAndLockForPut()返回的是null，会进入这个if语句，不需要新建HashEntry节点
                        oldValue = e.value;
                        if (!onlyIfAbsent) {
        
                            e.value = value;
                            ++modCount;
                        }
                        break;
                    }
                    e = e.next;
                }
                else {
        
                    if (node != null)
                        node.setNext(first);	// 如果在scanAndLockForPut()返回了一个HashEntry，则将该HashEntry作为链表首节点
                    else
                        node = new HashEntry<K,V>(hash, key, value, first);	// 如果tryLock()成功，则新建一个HashEntry将其作为链表首节点
                    int c = count + 1;
                    if (c > threshold && tab.length < MAXIMUM_CAPACITY)
                        rehash(node);	// 扩容操作
                    else
                        setEntryAt(tab, index, node);
                    ++modCount;
                    count = c;
                    oldValue = null;
                    break;
                }
            }
        } finally {
        
            unlock();
        }
        return oldValue;
    }

在put()方法中我们可以看到，如果tryLock()成功，不会进入scanAndLockForPut()方法自旋等待。一旦进入scanAndLockForPut()方法自旋等待，无论返回结果是否为null，都不影响后续代码逻辑。如果scanAndLockForPut()返回结果不为null，说明key不存在，会将该HashEntry置为链表首节点；如果scanAndLockForPut()返回结果为null，则不需要新建HashEntry节点，修改原来的节点对应的value值即可。

#### 3.1.2.2 rehash()函数 ####

在put()函数中我们看到，当新加入的节点后的节点数目大于threshold时，table数组会进行扩容操作，即rehash()函数：

private void rehash(HashEntry<K,V> node) {
        
        HashEntry<K,V>[] oldTable = table;
        int oldCapacity = oldTable.length;
        int newCapacity = oldCapacity << 1;
        threshold = (int)(newCapacity * loadFactor);
        HashEntry<K,V>[] newTable =
            (HashEntry<K,V>[]) new HashEntry[newCapacity];
        int sizeMask = newCapacity - 1;
        for (int i = 0; i < oldCapacity ; i++) {
        
            HashEntry<K,V> e = oldTable[i];
            if (e != null) {
        
                HashEntry<K,V> next = e.next;
                int idx = e.hash & sizeMask;
                if (next == null)   //  Single node on list
                    newTable[idx] = e;
                else {
         // Reuse consecutive sequence at same slot
                    HashEntry<K,V> lastRun = e;
                    int lastIdx = idx;
                    for (HashEntry<K,V> last = next;
                         last != null;
                         last = last.next) {
        
                        int k = last.hash & sizeMask;
                        if (k != lastIdx) {
        
                            lastIdx = k;
                            lastRun = last;
                        }
                    }
                    newTable[lastIdx] = lastRun;
                    // Clone remaining nodes
                    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);
                    }
                }
            }
        }
        int nodeIndex = node.hash & sizeMask; // add the new node
        node.setNext(newTable[nodeIndex]);
        newTable[nodeIndex] = node;
        table = newTable;
    }

总结一下上述代码的逻辑：

*  新数组的大小为原table数组的2倍。
 *  将原table数组中的节点全部迁移至新数组中，由于rehash()时已经获取到锁，不用考虑多线程情况下的线程安全问题。这个迁移操作和**2.1.3.2 扩容**中的迁移操作不同，这个迁移操作会新建部分HashEntry节点：对于链表末尾处某段连续的节点，这段节点全部迁移至新数组同一个下标处，这部分节点会一起迁移；而其余节点则会被克隆，采用头插法，一个一个地迁移至新数组，在默认loadFactor等于0.75的情况下，这部分需要克隆的节点约占1/6。

##### 3.1.2.2.1 为什么需要克隆HashEntry节点？为什么仅需要克隆部分HashEntry节点？ #####

ConcurrentHashMap的读操作是不需要加锁的，当我们在rehash()过程中，可能正有其他线程在访问原table数组，如果我们改变原table数组中HashEntry的next指针，可能会导致这些读线程访问错乱，因此需要克隆HashEntry节点。而对于那些位于链表末尾处的某段连续的HashEntry链表，这段链表中的节点全部迁移至新数组同一个下标处，其next指针不需要发生任何改变，因此不需要克隆，这是为了提高性能。

### 3.1.3 构造函数 ###

public ConcurrentHashMap(int initialCapacity,
                             float loadFactor, int concurrencyLevel) {
        
        // 如果输入参数非法，直接抛出IllegalArgumentException异常
        if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0)
            throw new IllegalArgumentException();
        // 如果并发级别concurrencyLevel超出了最大并发数限制，将并发级别设置为最大并发数MAX_SEGMENTS
        if (concurrencyLevel > MAX_SEGMENTS)
            concurrencyLevel = MAX_SEGMENTS;
        // Find power-of-two sizes best matching arguments
        int sshift = 0;
        // ssize表示Segment数组的大小，必须大于等于期望的并发级别concurrencyLevel, 且ssize必须是2的整数次幂
        int ssize = 1;
        while (ssize < concurrencyLevel) {
        
            ++sshift;
            ssize <<= 1;
        }
        this.segmentShift = 32 - sshift;
        // segmentMask用于确定key位于Segment数组的哪个位置
        this.segmentMask = ssize - 1;
        if (initialCapacity > MAXIMUM_CAPACITY)
            initialCapacity = MAXIMUM_CAPACITY;
        // c表示一个Segment中至少需要多少个桶
        int c = initialCapacity / ssize;
        if (c * ssize < initialCapacity)
            ++c;
        // cap是一个Segment中的哈希桶数目，必须大于等于c，且是2的整数次幂
        int cap = MIN_SEGMENT_TABLE_CAPACITY;
        while (cap < c)
            cap <<= 1;
        // create segments and segments[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];
        UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]
        this.segments = ss;
    }

经过上述构造函数，下述3个成员变量被赋值：

// segmentMask和segmentShift用于根据给定的key的hash值定位到Segment的位置
    final int segmentMask;  
    final int segmentShift;
    // 经过构造函数，segments的第0号位置已经被初始化，其余位置未被初始化
    final Segment<K,V>[] segments;

#### 3.1.3.1 为什么在构造函数中仅初始化下标为0的Segment？ ####

主要是为了性能上的考虑，其余的Segment只会在被使用到的时候才会调用ensureSegment()方法进行初始化，该方法如下：

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<K,V> proto = ss[0]; // use segment 0 as prototype
            int cap = proto.table.length;
            float lf = proto.loadFactor;
            int threshold = (int)(cap * lf);
            HashEntry<K,V>[] tab = (HashEntry<K,V>[])new HashEntry[cap];
            if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
                == null) {
         // recheck
                Segment<K,V> s = new Segment<K,V>(lf, threshold, tab);
                while ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
                       == null) {
        
                    if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s))
                        break;
                }
            }
        }
        return seg;
    }

总结上述方法的逻辑如下：

*  k指的是我们需要初始化segments数组中第k个元素。
 *  新的Segment是从segments\[0\]复制而来的，并使用CAS对segments\[k\]赋值，确保线程安全性。

#### 3.1.3.2 为什么在构造函数中调用UNSAFE.putOrderedObject()方法给segments数组的0号下标初始化？ ####

先来看一下UNSAFE.putObjectVolatile()和UNSAFE.putOrderedObject()的区别：

// 存储变量的引用到对象的指定的偏移量处，使用volatile的存储语义
    public native void putObjectVolatile(Object o, long offset, Object x);
    // 有序、延迟版本的putObjectVolatile方法，不保证值的改变被其他线程立即看到。
    public native void putOrderedObject(Object o, long offset, Object x);

再来看看Segment数组segments的定义：

final Segment<K,V>[] segments;

segments是一个final修饰的属性，编译器会在final域的写之后，构造函数return之前，插入一个StoreStore屏障，这个屏障保证了segments数组在构造函数外的可见性，因此我们仅需调用UNSAFE.putOrderedObject()，而无需调用更耗费性能的UNSAFE.putObjectVolatile()方法。

### 3.1.4 lazySet ###

由UNSAFE.putOrderedObject()的内存语义可知，该操作不保证可见性，但是如果其后面紧跟着unlock()操作，就可以由unlock()操作来保证内存可见性，无需调用更耗费性能的UNSAFE.putOrderedObject()方法。这就是ConcurrentHashMap中所谓的lazySet的概念，即调用UNSAFE.putOrderedObject()方法填充的属性只有在unlock()操作之后才能保证可见性。

### 3.1.5 聚合函数的实现 ###

在多线程环境下，有许多线程在访问各个不同的Segment，如何实现一些聚合函数呢？比如size()。核心思想是：首先尝试着不加锁统计size()的数目，如果统计不成功，再重新统计。如果经过若干次尝试后依然失败，则对所有Segment加锁再统计。

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 (;;) {
        
                // 重试次数过多，直接对segments全部加锁统计size
                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;
                    }
                }
                // 只有本次modCount总数和上一次modCount总数相同时，说明没有发生结构性修改，才会跳出循环
                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;
    }

### 3.1.6 总结 ###

将table数组绘制成一张图：

![img][img 2]

JDK1.7的ConcurrentHashMap源码总共1620行，采用数组+链表+分段锁实现，并采用了大量移位操作提高运算速度。

## 3.2 JDK1.8中的实现 ##

### 3.2.1 设计思想 ###

*  和HashMap一样，在JDK1.7ConcurrentHashMap的基础上引入了红黑树这一数据结构来优化哈希冲突下的性能。
 *  抛弃了Segment的概念，但依然采用分段锁的思想，但是锁粒度不再是Segment，而是每一个数组下标中的元素，即锁粒度更细。
 *  不采用ReentrantLock加锁，直接采用原生的synchronized关键字加锁。
 *  采用一种全新的TableStack数据结构，用基于栈的思想解决多线程环境下的遍历问题。因此在扩容操作迁移节点时，我们不需要克隆节点，直接操作原节点即可。
 *  为lambda表达式和Spliterator提供了方法和类的支持。

### 3.2.2 节点定义 ###

*  Node：最普通的链表节点。
 *  TreeNode：红黑树节点。
 *  ForwardingNode：扩容操作迁移节点时的转发节点，hash值为MOVED=-1。
 *  ReservationNode：compute()和computeIfAbsent()方法使用的标记节点，hash值为RESERVED=-3。
 *  TreeBin：持有红黑树对象的节点，hash值为TREEBIN=-2。相比于JDK1.8的HashMap，TreeBin的引入是因为在红黑树的插入和删除节点过程中，红黑树的根节点会发生变化，不便于对根节点进行synchronized加锁。

TreeNode、ForwardingNode、ReservationNode和TreeBin都是Node的子类。

### 3.2.3 sizeCtl的含义 ###

/**
     * Table initialization and resizing control.  When negative, the
     * table is being initialized or resized: -1 for initialization,
     * else -(1 + the number of active resizing threads).  Otherwise,
     * when table is null, holds the initial table size to use upon
     * creation, or 0 for default. After initialization, holds the
     * next element count value upon which to resize the table.
     */
    private transient volatile int sizeCtl;

sizeCtl是ConcurrentHashMap的一个属性，用volatile关键词修饰，用于控制table数组初始化和扩容操作。根据注释，sizeCtl可分为以下几种含义：

*  当sizeCrl为-1时，表明table数组正在被初始化。
 *  当sizeCtl为其他负数时，表明table数组正在扩容，且其值为-(1 + 正在参与扩容的线程数量)。
 *  当sizeCtl为0时，表明此时table数组为null，使用默认的初始化大小，即16。
 *  当sizeCtl为正数，且table数组为null时，sizeCtl代表table的初始化大小。
 *  当sizeCtl为正数，且table数组不为null时，sizeCtl代表下一次扩容的扩容阈值。

实际上，上述第2点是有误的，正确的描述应该是：

当sizeCtl为其他负数时，其高x位代表此次扩容操作数组的一个标记（与table数组的大小有关），低(32-x)位代表(1+正在参与扩容的线程数)，其中x的值为RESIZE\_STAMP\_BITS，在ConcurrentHashMap的实现中是一个常量16。

### 3.2.4 哈希函数 ###

static final int spread(int h) {
        
        return (h ^ (h >>> 16)) & HASH_BITS;
    }

相比于**2.2.3 哈希函数**中提及的JDK1.8中HashMap的哈希函数，多了和HASH\_BITS（默认值是0x7fffffff）的位与运算。由3.2.2 节点定义可知，在JDK1.8版本的ConcurrentHashMap中有哈希值为负数的节点，HASH\_BITS正是为了将符号位置为0，排除这些节点的哈希值。

### 3.2.5 put()函数 ###

final V putVal(K key, V value, boolean onlyIfAbsent) {
        
        // 对于ConcurrentHashMap，key和value都不能为null
        if (key == null || value == null) throw new NullPointerException();
        // spread()方法就是HashMap中的hash()方法
        int hash = spread(key.hashCode());
        // binCount用于记录ConcurrentHashMap中的键值对数目
        int binCount = 0;
        for (Node<K,V>[] tab = table;;) {
        
            Node<K,V> f; int n, i, fh;
            if (tab == null || (n = tab.length) == 0)
                // 如果table数组 还没有被初始化，则先初始化table数组
                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
                }
            // 如果遇到了ForwardingNode，则先帮助迁移节点，再在下一次进入循环时放置值
            else if ((fh = f.hash) == MOVED)
                tab = helpTransfer(tab, f);
            else {
        
                V oldVal = null;
                // synchronized对table数组中第i位置处的元素f加锁
                synchronized (f) {
        
                    if (tabAt(tab, i) == f) {
        
                        if (fh >= 0) {
        
                        	// 如果是链表节点，插入链表末尾，binCount是计数用的，记录插入新节点前链表中的节点数
                            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)
                    	// 如果插入新节点前，链表中的节点数大于等于 TREEIFY_THRESHOLD，则树化
                        treeifyBin(tab, i);
                    if (oldVal != null)
                        return oldVal;
                    break;
                }
            }
        }
        addCount(1L, binCount);
        return null;
    }

put()函数的基本思想可以总结如下：

*  根据传入key的哈希值确定数组下标i。
 *  如果是第一次调用put()函数，即table数组还没有被初始化，则初始化table数组，这个过程在initTable()函数中。
 *  如果table\[i\]为null，利用CAS操作将table\[i\]置为新Node节点。
 *  如果table\[i\]为ForwardingNode，则参与转移节点，这个过程在helpTransfer()函数中。
 *  否则对table\[i\]加synchronized锁，如果table\[i\]是TreeBin，则往该TreeBin持有的红黑树中添加新节点；如果table\[i\]只是普通的Node节点，往链表末尾添加，如果插入新节点前，链表中的节点数大于等于 TREEIFY\_THRESHOLD，则树化。
 *  调用addCount()方法将节点数加1。

#### 3.2.5.1 initTable()函数 ####

private final Node<K,V>[] initTable() {
        
        Node<K,V>[] tab; int sc;
        while ((tab = table) == null || tab.length == 0) {
        
            if ((sc = sizeCtl) < 0)
            	// 如果sizeCtl小于0，说明已经有线程在初始化table数组，当前线程yield()，让出CPU
                Thread.yield(); // lost initialization race; just spin
            // CAS操作将sizeCtl变量置为-1，表示table数组正在被初始化
            else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
        
                try {
        
                    if ((tab = table) == null || tab.length == 0) {
        
                    	// 如果sc大于0，则将sc作为table数组初始容量；否则，将table数组初始容量设置为16
                        int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
                        Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                        table = tab = nt;
                        // sc存储下一次扩容阈值，是0.75n，采用位运算实现
                        sc = n - (n >>> 2);
                    }
                } finally {
        
                	// 将sizeCtl重置为下一次扩容阈值
                    sizeCtl = sc;
                }
                break;
            }
        }
        return tab;
    }

initTable()函数考虑了多线程访问情况下的线程安全性，总结一下这个函数的逻辑：

只要table数组没有被初始化，则一直循环。在循环过程中，如果发现sizeCtl小于0，这说明已经有现车在初始化table数组，当前线程调用Thread.yield()方法主动让出CPU时间片。如果sizeCtl大于等于0，尝试使用CAS操作将sizeCtl置为-1，向其他线程表明当前线程正在初始化数组。

#### 3.2.5.2 helpTransfer()函数 ####

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);	// 当tab.length为16时，rs为32795
            while (nextTab == nextTable && table == tab &&
                   (sc = sizeCtl) < 0) {
        
                if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
                    sc == rs + MAX_RESIZERS || transferIndex <= 0)
                    break;
                // CAS操作设置正在扩容的线程数加1
                if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
        
                    transfer(tab, nextTab);	// 参与扩容
                    break;
                }
            }
            return nextTab;
        }
        return table;
    }

helpTransfer()函数的逻辑总结如下：

确认传入的参数f节点是一个ForwardingNode节点，且其nextTable属性不为null，表明当前table数组正处于转移节点的过程中。调用resizeStamp()函数计算出一个戳标记rs，该戳标记是根据table数组的长度生成的，作为本次扩容操作的一个戳标记。接着会进入一个while循环，只要转移节点的过程没有完成，且满足某种特定条件（即下文分析的条件一~条件四），就会利用CAS操作设置sizeCtl的值加1，表明参与转移节点的线程数增加1。

##### 3.2.5.2.1 一个小错误 #####

在helpTransfer()方法和addCount()方法中，出现过下述代码：

// 条件一
    (sc >>> RESIZE_STAMP_SHIFT) != rs || 
    // 条件二
    sc == rs + 1 || 
    // 条件三
    sc == rs + MAX_RESIZERS || 
    // 条件四
    transferIndex <= 0

条件一~条件四分别代表什么含义呢？

从helpTransfer()方法和addCount()方法这两个调用点中我们可以看到两点：

*  判断这4个条件时，sc肯定是一个负数。
 *  当第一个线程参与扩容时，sc的值被置为(rs << RESIZE\_STAMP\_SHIFT) + 2。

###### 3.2.5.2.1.1 条件一：(sc >>> RESIZE\_STAMP\_SHIFT) != rs ######

其中，sc代表的是sizeCtl的值，RESIZE\_STAMP\_SHIFT = 32 - RESIZE\_STAMP\_BITS=16。

rs由下述函数计算所得：

static final int resizeStamp(int n) {
        
        return Integer.numberOfLeadingZeros(n) | (1 << (RESIZE_STAMP_BITS - 1));
    }

rs代表的就是此次扩容操作table数组的一个标记（只与table数组的大小有关），在RESIZE\_STAMP\_BITS为16的情况下，rs一定是个正数。

此时sizeCtl为负数，且其高16位与rs不相等，说明此次扩容已经结束（因为扩容戳标记rs已经发生改变）。

###### 3.2.5.2.1.2 条件二：sc == rs + 1 ######

由前述分析可知，能走到这个条件时，sc一定是个负数，rs一定是个正数，这个条件在正常的ConcurrentHashMap代码逻辑里是永远不可能满足的！

###### 3.2.5.2.1.3 条件三：sc == rs + MAX\_RESIZERS ######

由条件二中的分析可知，sc是负数，rs是正数，这种情况在正常的ConcurrentHashMap代码里也是永远不可能出现的。

###### 3.2.5.2.1.4 条件四：transferIndex <= 0 ######

这个条件比较好理解，即原table数组里的所有元素都已经有线程在负责迁移至新的table数组里，本线程无需再继续参与扩容。（后文会对transferIndex的含义做分析）

既然条件二和条件三永远无法满足，为什么还要这么写呢？

真相就是这是JDK1.8的一个bug，正确的写法应该是：

条件二：sc == (rs << RESIZE\_STAMP\_SHIFT) +1，此时代表没有任何线程在进行扩容操作，当前线程无需再继续参与扩容。

条件三：sc == (rs << RESIZE\_STAMP\_SHIFT) + MAX\_RESIZERS，此时进行扩容操作的线程数达到了MAX\_RESIZERS - 1，当前线程无法再继续参与扩容。

该bug在JDK12中被修复，证据链接：

https://bugs.java.com/bugdatabase/view\_bug.do?bug\_id=JDK-8214427

##### 3.2.5.2.2 transfer()函数 #####

private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
        
        int n = tab.length, stride;
        // 计算每个线程参与移动的哈希桶数量stride
        if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
            stride = MIN_TRANSFER_STRIDE; // subdivide range
        // 如果nextTab还没有被初始化，先初始化nextTab
        if (nextTab == null) {
                    // initiating
            try {
        
            		// 每次扩容数组容量乘以2
                Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
                nextTab = nt;
            } catch (Throwable ex) {
              // try to cope with OOME
            		// 如果发生OOM 异常，说明无法再继续扩容，设置扩容阈值sizeCtl为Integer.MAX_VALUE，防止下次再进到扩容逻辑里
                sizeCtl = Integer.MAX_VALUE;
                return;
            }
            nextTable = nextTab;
            // transferIndex 是一个volatile类型的变量，用以标识当前线程从哪个桶开始迁移
            transferIndex = n;
        }
        int nextn = nextTab.length;
        ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
        // 迁移原数组中的元素时，从数组末往数组首遍历，advance用以标识当前线程是否可以去当
        // 前数组左侧寻找待迁移的桶，finishing代表迁移是否已经结束
        boolean advance = true;
        boolean finishing = false; // to ensure sweep before committing nextTab
        for (int i = 0, bound = 0;;) {
        
            Node<K,V> f; int fh;
            while (advance) {
        
                int nextIndex, nextBound;
                // 当i仍然在遍历当前线程需要迁移的桶，或者当前迁移已经结束时，无需去当前数组的左侧寻找待迁移的桶
                if (--i >= bound || finishing)
                    advance = false;
                // 如果待迁移的第一个桶已经小于等于0，说明所有的桶都有线程在负责迁移，无需去当前数组的左侧继续寻找待迁移的桶
                else if ((nextIndex = transferIndex) <= 0) {
        
                    i = -1;
                    advance = false;
                }
                // 用CAS操作设置当前线程负责迁移的桶号
                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;
                    sizeCtl = (n << 1) - (n >>> 1);
                    return;
                }
                // 用CAS操作设置sizeCtl减1
                if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
        
                		// 当n等于16时，(resizeStamp(16)) << 16 + 2 = -2145714174
                		// 表示成二进制是：10000000 00011011 00000000 00000010，其高
                		// 16位的值和resizeStamp(16)恰好相等；其低位为2，表示现在只
                		// 有1个线程在帮助扩容。如果不是只有1个线程在扩容，没必要去检查扩容
                		// 是否完成，直接return即可。
                    if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
                        return;
                    finishing = advance = true;
                    i = n; // recheck before commit
                }
            }
            else if ((f = tabAt(tab, i)) == null)
            		// CAS 操作设置位置i的节点为fwd，即该位置已迁移完毕。
            		// 如果设置成功，则advance为true，表明可以去寻找下一个待迁移的位置。
            		// 如果设置失败，则advance为false，表明不能去寻找下一个待迁移的位置
              	// 注意外层是一个while死循环，因此即使当前设置失败，还会循环进入到这里继续CAS尝试，直到成功为止。
                advance = casTabAt(tab, i, null, fwd);
            else if ((fh = f.hash) == MOVED)
                advance = true; // already processed
            else {
        
            		// synchronized加锁后开始迁移节点
                synchronized (f) {
        
                    if (tabAt(tab, i) == f) {
        
                        Node<K,V> ln, hn;
                        if (fh >= 0) {
        
                        		// fh 大于等于 0，说明是正常的链表节点
                        		// 这个过程在JDK1.7的ConcurrentHashMap 也出现过，先把末尾部分将移至
                        		// 新数组的节点串在一起转移，再遍历前面的节点一个个转移到两个链表上，最后
                        		// 再将两个链表挂接在新数组中。 
                            int runBit = fh & n;
                            Node<K,V> lastRun = f;
                            for (Node<K,V> p = f.next; p != null; p = p.next) {
        
                                int b = p.hash & n;
                                if (b != runBit) {
        
                                    runBit = b;
                                    lastRun = p;
                                }
                            }
                            if (runBit == 0) {
        
                                ln = lastRun;
                                hn = null;
                            }
                            else {
        
                                hn = lastRun;
                                ln = null;
                            }
                          	// 这里为什么要克隆Node节点？在3.1.2.2.1 为什么需要克隆HashEntry节点？为什么仅需要克隆
                          	// 部分HashEntry节点？中有解释，是相同的原因。
                            for (Node<K,V> p = f; p != lastRun; p = p.next) {
        
                                int ph = p.hash; K pk = p.key; V pv = p.val;
                                if ((ph & n) == 0)
                                    ln = new Node<K,V>(ph, pk, pv, ln);
                                else
                                    hn = new Node<K,V>(ph, pk, pv, hn);
                            }
                            setTabAt(nextTab, i, ln);
                            setTabAt(nextTab, i + n, hn);
                            setTabAt(tab, i, fwd);
                            advance = true;
                        }
                        else if (f instanceof TreeBin) {
        
                        		// 对于树节点，先根据next指针遍历原table数组中的节点，并
                        		// 利用其在新table数组中的位置将其分成两个链表，再将新的两个
                        		// 链表挂接至新table数组中。
                            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;
                                    else
                                        loTail.next = p;
                                    loTail = p;
                                    ++lc;
                                }
                                else {
        
                                    if ((p.prev = hiTail) == null)
                                        hi = p;
                                    else
                                        hiTail.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;
                        }
                    }
                }
            }
        }
    }

总结一下transfer()函数的逻辑：

首先明确transferIndex的概念，其定义如下：

private transient volatile int transferIndex;

可以看到，transferIndex是ConcurrentHashMap中的一个volatile类型的成员变量，确保了多个线程访问时的内存可见性。每一个线程参与转移节点的过程，负责转移的哈希桶数目都是确定的，假如每个线程负责转移的哈希桶数目是16个，那么当前线程负责转移的哈希桶下标就是\[transferIndex - 16, transferIndex - 1\]，并更新transferIndex的值为transferIndex - 16。

在转移每个哈希桶时，都会给这个哈希桶首节点加synchronized锁（对于红黑树，是给TreeBin加锁），转移节点的过程不复杂，对于链表节点的转移，和JDK1.7版本的ConcurrentHashMap是一致的；对于红黑树节点的转移，先将其分成两个链表，在分别判断是否需要去树化，这和JDK1.8版本的HashMap是一致的。

#### 3.2.5.3 addCount()函数 ####

private final void addCount(long x, int check) {
        
        CounterCell[] as; long b, s;
        if ((as = counterCells) != null ||
            !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
        
        	// 如果counterCells数组不为null，或者说CAS尝试修改 baseCount 失败
            CounterCell a; long v; int m;
            boolean uncontended = true;
            if (as == null || (m = as.length - 1) < 0 ||
                (a = as[ThreadLocalRandom.getProbe() & m]) == null ||
                !(uncontended =
                  U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
        
                // 该方法和Striped64的longAccumulate()方法一模一样
                fullAddCount(x, uncontended);
                return;
            }
            if (check <= 1)
            	// check <= 1，说明不是增加节点的操作，无需扩容，因此直接 return。
                return;
            // sumCount()将baseCount和counterCells中存储的值都加在一起
            s = sumCount();
        }
        if (check >= 0) {
        
            Node<K,V>[] tab, nt; int n, sc;
            // 如果元素数大于等于sizeCtl，且table数组不为null，且table数组容量小于MAXIMUM_CAPACITY，则需要进行扩容操作
            while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
                   (n = tab.length) < MAXIMUM_CAPACITY) {
        
                int rs = resizeStamp(n);
                if (sc < 0) {
        
                    if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
                        sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
                        transferIndex <= 0)
                        break;
                    if (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();
            }
        }
    }

addCount()方法是记录ConcurrentHashMap中的键值对数目的，将键值对数目存放在baseCount和counterCells中，这两个属性的定义如下：

private transient volatile long baseCount;
    private transient volatile CounterCell[] counterCells;

如果CAS设置baseCount失败，则随机获取一个counterCells中某下标位置的synchronized锁，将该下标位置的计数加1。这其实是分段锁思想的提现，在sumCount()中我们会将baseCount和所有counterCells中的计数值都加在一起来统计ConcurrentHashMap中的键值对数目：

final long sumCount() {
        
        CounterCell[] as = counterCells; CounterCell a;
        long sum = baseCount;
        if (as != null) {
        
            for (int i = 0; i < as.length; ++i) {
        
                if ((a = as[i]) != null)
                    sum += a.value;
            }
        }
        return sum;
    }

值得注意的是，size()函数也是直接调用sumCount()函数实现的：

public int size() {
        
        long n = sumCount();
        return ((n < 0L) ? 0 :
                (n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :
                (int)n);
    }

而sumCount()函数没有加锁，因此在多线程环境下通过size()函数得到的键值对数目是不准确的。

addCount()函数不仅在putVal()函数中被调用，在remove()函数等其他位置也会被调用，如果是在putVal()中被调用，其check为binCount，是大于0的一个值，会进入扩容逻辑的判断中。

### 3.2.6 get()函数 ###

public V get(Object key) {
        
        Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
        // spread()函数对应HashMap中的hash()函数
        int h = spread(key.hashCode());
        if ((tab = table) != null && (n = tab.length) > 0 &&
            (e = tabAt(tab, (n - 1) & h)) != null) {
        
            if ((eh = e.hash) == h) {
        
                if ((ek = e.key) == key || (ek != null && key.equals(ek)))
                    return e.val;
            }
            else if (eh < 0)
            		// 有3种节点的hash值是小于0的，TreeBin、ForwardingNode和ReservationNode，这三个类均继承自Node，重写了find()方法
                return (p = e.find(h, key)) != null ? p.val : null;
            //走到这，说明是链表，遍历链表寻找
            while ((e = e.next) != null) {
        
                if (e.hash == h &&
                    ((ek = e.key) == key || (ek != null && key.equals(ek))))
                    return e.val;
            }
        }
        return null;
    }

get()函数是不需要加锁的，对于哈希值为负数的节点调用其特有的find()方法去查找；对于哈希值为正数的链表节点，遍历链表查找。

#### 3.2.6.1 TreeBin中的find()函数 ####

TreeBin是自带读写锁的节点，来看看其中的2个属性：

volatile Thread waiter;
    volatile int lockState;

读线程是不需要加锁的，写线程是需要获取独占锁的，waiter表示当前正在等待获取独占锁的写线程，lockState表示当前的锁状态。

final Node<K,V> find(int h, Object k) {
        
        if (k != null) {
        
            for (Node<K,V> e = first; e != null; ) {
        
                int s; K ek;
                // 如果当前节点的状态中有WAITER和WRITER状态，说明树在重构中。
                // 则根据next指针，以遍历链表的方式来遍历树，来寻找节点。（这样
                // 做比在红黑树中寻找节点要低效些，但不至于阻塞）
                if (((s = lockState) & (WAITER|WRITER)) != 0) {
        
                    if (e.hash == h &&
                        ((ek = e.key) == k || (ek != null && k.equals(ek))))
                        return e;
                    e = e.next;
                }
                // 如果当前节点的状态只有READER，则CAS获取到锁
                else if (U.compareAndSwapInt(this, LOCKSTATE, s,
                                             s + READER)) {
        
                    TreeNode<K,V> r, p;
                    try {
        
                    		// 用TreeNode的find()方法来寻找节点
                        p = ((r = root) == null ? null :
                             r.findTreeNode(h, k, null));
                    } finally {
        
                        Thread w;
                        // 重置变量lockState，如果此时既有WAITER状态也有READER状态，
                        // 但是没有WRITER 状态，则唤醒等待线程waiter。
                        if (U.getAndAddInt(this, LOCKSTATE, -READER) ==
                            (READER|WAITER) && (w = waiter) != null)
                            LockSupport.unpark(w);
                    }
                    return p;
                }
            }
        }
        return null;
    }

总结上述find()函数的逻辑：

如果有线程获取了写锁或者正在等待获取写锁，读线程不会参与竞争锁状态，会沿着节点的next指针遍历寻找节点。沿着链表的next指针寻找节点虽然比较慢，但至少不会阻塞。

否则，利用CAS操作将读线程数量加1，如果CAS成功则调用红黑树的find()函数查找节点，这是O(logn)的时间复杂度。

#### 3.2.6.2 ForwardingNode中的find()函数 ####

Node<K,V> find(int h, Object k) {
        
        // loop to avoid arbitrarily deep recursion on forwarding nodes
        // 在ForwardingNode节点指向的下一个数组nextTable中寻找节点。死循环是
        // 为了避免扩容操作导致的当前ForwardingNode指向的数组不是最终数组。
        outer: for (Node<K,V>[] tab = nextTable;;) {
        
            Node<K,V> e; int n;
            if (k == null || tab == null || (n = tab.length) == 0 ||
                (e = tabAt(tab, (n - 1) & h)) == null)
                return null;
            for (;;) {
        
                int eh; K ek;
                if ((eh = e.hash) == h &&
                    ((ek = e.key) == k || (ek != null && k.equals(ek))))
                    return e;
                if (eh < 0) {
        
                    if (e instanceof ForwardingNode) {
        
                        tab = ((ForwardingNode<K,V>)e).nextTable;
                        continue outer;
                    }
                    else
                        // 递归调用，此时的e可能是TreeBin、Node、TreeNode、ReservationNode中的任何一个。
                        return e.find(h, k);
                }
                if ((e = e.next) == null)
                    return null;
            }
        }
    }

总结一下上述函数的逻辑：

ForwardingNode节点是一个转发节点，因此只负责将find()操作转移至nextTable属性中去查找。

#### 3.2.6.3 ReservationNode中的find()函数 ####

Node<K,V> find(int h, Object k) {
        
        return null;
    }

ReservationNode()仅在computeIfAbsent()等函数中作为标记节点使用，我们不会调用其find()函数。

#### 3.2.6.4 Node中的find()函数 ####

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;
    }

### 3.2.7 迭代器的实现 ###

#### 3.2.7.1 设计思想 ####

在迭代器遍历的过程中，可能有线程正在对当前ConcurrentHashMap进行扩容操作，因此会出现部分节点在原table数组，部分节点在新table数组中的情况。ConcurrentHashMap基于一种TrackStack的栈节点构成一个栈。当出现数组扩容时转移节点时，如果遇到了ForwardingNode节点，会将当前数组的遍历状态（包括当前数组遍历到的索引index、当前数组的大小baseSize、当前数组的遍历边界baseLimit）压入栈中，并在新table数组的索引index和index + baseSize处分别遍历这两个桶中的节点。在遍历完新数组中的这两个桶之后，再从栈中取出原table数组的状态继续遍历第index + 1位置。

#### 3.2.7.2 类结构 ####

\[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-u7flBsq4-1612106552359)(/Users/qyh/Library/Application Support/typora-user-images/image-20210131154445226.png)\]

由上图可知，所有的迭代器都基于Traverser实现，来看一下Traverser中的属性：

Node<K,V>[] tab;        // current table; updated if resized
    Node<K,V> next;         // the next entry to use
    // stack代表着我们需要使用的栈，spare中存储的是一些废弃不用的TrackStack对象，需要使用的时候可以从spare中获取，而无需新建
    TableStack<K,V> stack, spare; // to save/restore on ForwardingNodes
    int index;              // index of bin to use next
    int baseIndex;          // current index of initial table
    int baseLimit;          // index bound for initial table
    final int baseSize;     // initial table size

TrackStack代表一个栈帧，其定义如下：

static final class TableStack<K,V> {
        
        int length;
        int index;
        Node<K,V>[] tab;
        TableStack<K,V> next;
    }

#### 3.2.7.3 缺陷 ####

利用栈结构实现迭代器有一个缺陷在于：假设当前遍历到了table数组中index位置，另外的线程往\[0, index - 1\]处添加了元素，那么这个新添加的元素将永远没有机会被遍历到。当然这个缺陷是被Doug Lea考虑到了的，是一种可以接受的缺陷。

### 3.2.8 总结 ###

将table数组绘制成一张图：

![img][img 3]

JDK1.8的ConcurrentHashMap代码对于volatile、synchronized、CAS等和Java内存与锁相关的知识运用得淋漓尽致，但也存在着些许小bug。

# 4. 写在最后 #

本文基本上总结了JDK1.7和JDK1.8中HashMap和ConcurrentHashMap的实现思想和细节，用以个人输出总结。虽然配图较少，但我觉得我还是把我知道的关键点都说清楚了，撒花。

[img]: /images/20221119/090cdb73a072439f9b9f6ebb233acc9c.png
[img 1]: /images/20221119/5affc2a33f9b4ac089f4ab96f8a8f3ee.png
[img 2]: /images/20221119/eaefcda6556a430da3a7e69068217974.png
[img 3]: /images/20221119/4d8f1031ba9c4db68f53a9fb0cd2c56a.png