转载

深入剖析 Java 7 中的 HashMap 和 ConcurrentHashMap

本文将深入剖析 Java7 中的 HashMap 和 ConcurrentHashMap 的源码，解析 HashMap 线程不安全的原理以及解决方案，最后以测试用例加以验证。

1 Java7 HashMap

HashMap 的数据结构：

从上图中可以看出，HashMap 底层就是一个数组结构，数组中的每一项又是一个链表。

通过查看 JDK 中的 HashMap 源码，可以看到其构造函数有一行代码：

public HashMap(int initialCapacity, float loadFactor) {
    ...
    table = new Entry[capacity];
    ...
}

即创建了一个大小为 capacity 的 Entry 数组，而 Entry 的结构如下：

static class Entry<K,V> implements Map.Entry<K,V> {
    final K key;
    V value;
    Entry<K,V> next;
    final int hash;
    ……
}

可以看到，Entry 是一个 static class，其中包含了 key 和 value ，也就是键值对，另外还包含了一个 next 的 Entry 指针。

capacity：当前数组容量，始终保持 2^n，可以扩容，扩容后数组大小为当前的 2 倍。默认初始容量为 16。
loadFactor：负载因子，默认为 0.75。
threshold：扩容的阈值，等于 capacity * loadFactor

1.1 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);
    // key 的 hash 值
    int hash = hash(key);
    // 找到对应的数组下标
    int i = indexFor(hash, table.length);
    // 遍历一下对应下标处的链表，看是否有重复的 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++;
    // 不存在重复的 key，将此 entry 添加到链表中
    addEntry(hash, key, value, i);
    return null;
}

这里对一些方法做深入解析。

数组初始化

private void inflateTable(int toSize) {
    // 保证数组大小一定是 2^n
    int capacity = roundUpToPowerOf2(toSize);
    // 计算扩容阈值
    threshold = (int) Math.min(capacity * loadFactor, MAXIMUM_CAPACITY + 1);
    // 初始化数组
    table = new Entry[capacity];
    initHashSeedAsNeeded(capacity);
}

找到对应的数组下标

static int indexFor(int hash, int length) {
    // 作用等价于取模运算，但这种方式效率更高
    return hash & (length-1);
}

因为HashMap的底层数组长度总是 2^n，当 length 为 2 的 n 次方时，hash & (length-1) 就相当于对length取模，而且速度比直接取模要快的多。

添加节点到链表中

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);
}
// 永远都是在链表的表头添加新元素
void createEntry(int hash, K key, V value, int bucketIndex) {
    // 获取指定 bucketIndex 索引处的 Entry
    Entry<K,V> e = table[bucketIndex];
    // 将新创建的 Entry 放入 bucketIndex 索引处，并让新的 Entry 指向原来的 Entry
    table[bucketIndex] = new Entry<>(hash, key, value, e);
    size++;
}

当系统决定存储 HashMap 中的 key-value 对时，完全没有考虑 Entry 中的 value，仅仅只是 根据 key 来计算并决定每个 Entry 的存储位置 。我们完全可以把 Map 集合中的 value 当成 key 的附属，当系统决定了 key 的存储位置之后，value 随之保存在那里即可。

数组扩容

随着 HashMap 中元素的数量越来越多，发生碰撞的概率将越来越大，所产生的子链长度就会越来越长，这样势必会影响 HashMap 的存取速度。为了保证 HashMap 的效率，系统必须要在某个临界点进行扩容处理，该临界点 threshold。而在 HashMap 数组扩容之后，最消耗性能的点就出现了：原数组中的数据必须重新计算其在新数组中的位置，并放进去，这就是 resize。

void resize(int newCapacity) {
    Entry[] oldTable = table;
    int oldCapacity = oldTable.length;
    // 若 oldCapacity 已达到最大值，直接将 threshold 设为 Integer.MAX_VALUE
    if (oldCapacity == MAXIMUM_CAPACITY) {
        threshold = Integer.MAX_VALUE;
        return; // 直接返回
    }
    // 否则，创建一个更大的数组
    Entry[] newTable = new Entry[newCapacity];
    //将每条Entry重新哈希到新的数组中
    transfer(newTable, initHashSeedAsNeeded(newCapacity));
    table = newTable;
    // 重新设定 threshold
    threshold = (int)Math.min(newCapacity * loadFactor, MAXIMUM_CAPACITY + 1);
}

1.2 get过程分析

public V get(Object key) {
    // key 为 null 的话，会被放到 table[0]，所以只要遍历下 table[0] 处的链表就可以了
    if (key == null)
        return getForNullKey();
    // key 非 null 的情况，详见下文
    Entry<K,V> entry = getEntry(key);
 
    return null == entry ? null : entry.getValue();
}
final Entry<K,V> getEntry(Object key) {
    // The number of key-value mappings contained in this map.
    if (size == 0) {
        return null;
    }
 
    // 根据该 key 的 hashCode 值计算它的 hash 码 
    int hash = (key == null) ? 0 : hash(key);
    // 确定数组下标，然后从头开始遍历链表，直到找到为止
    for (Entry<K,V> e = table[indexFor(hash, table.length)];
         e != null;
         e = e.next) {
        Object k;
        //若搜索的key与查找的key相同，则返回相对应的value
        if (e.hash == hash &&
            ((k = e.key) == key || (key != null && key.equals(k))))
            return e;
    }
    return null;
}

2 Java7 ConcurrentHashMap

ConcurrentHashMap 的成员变量中，包含了一个 Segment 数组 final Segment<K,V>[] segments;，而 Segment 是ConcurrentHashMap 的内部类。

然后在 Segment 这个类中，包含了一个 HashEntry 的数组transient volatile HashEntry<K,V>[] table，而 HashEntry 也是 ConcurrentHashMap 的内部类。

HashEntry 中，包含了 key 和 value 以及 next 指针（类似于 HashMap 中的 Entry），所以 HashEntry 可以构成一个链表。

2.1 成员变量及构造函数

public class ConcurrentHashMap<K, V> extends AbstractMap<K, V>
        implements ConcurrentMap<K, V>, Serializable { 
    ...
    //初始的容量
    static final int DEFAULT_INITIAL_CAPACITY = 16;
    //初始的加载因子
    static final float DEFAULT_LOAD_FACTOR = 0.75f;
    //初始的并发等级，表示当前更新线程的估计数
    static final int DEFAULT_CONCURRENCY_LEVEL = 16;
    //最大容量
    static final int MAXIMUM_CAPACITY = 1 << 30;
    //最小的segment数量
    static final int MIN_SEGMENT_TABLE_CAPACITY = 2;
    //最大的segment数量
    static final int MAX_SEGMENTS = 1 << 16; 
    //
    static final int RETRIES_BEFORE_LOCK = 2;
    // segments 的掩码值, key 的散列码的高位用来选择具体的 segment
    final int segmentMask; 
    // 偏移量
    final int segmentShift; 
    final Segment<K,V>[] segments; 
    ...
    // 创建一个带有指定初始容量、加载因子和并发级别的新的空映射
    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;
        // 寻找最佳匹配参数（不小于给定参数的最接近的 2^n）
        int sshift = 0; // 用来记录向左按位移动的次数
        int ssize = 1; // 用来记录Segment数组的大小
        // 计算并行级别 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;
        // 记录每个 Segment 上要放置多少个元素
        int c = initialCapacity / ssize;
        // 假如有余数，则Segment数量加1
        if (c * ssize < initialCapacity)
            ++c;
        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;
    }

当用 new ConcurrentHashMap() 无参构造函数进行初始化的，那么初始化完成后：

Segment 数组长度为 16，不可以扩容
Segment[i] 的默认大小为 2，负载因子是 0.75，得出初始阈值为 1.5，也就是以后插入第一个元素不会触发扩容，插入第二个会进行第一次扩容
这里初始化了 segment[0]，其他位置还是 null，至于为什么要初始化 segment[0]，后面的代码会介绍
当前 segmentShift 的值为 32 – 4 = 28，segmentMask 为 16 – 1 = 15，姑且把它们简单翻译为移位数和掩码，这两个值马上就会用到

2.2 put过程分析

根据 hash 值很快就能找到相应的 Segment，之后就是 Segment 内部的 put 操作。

public V put(K key, V value) {
    Segment<K,V> s;
    if (value == null)
        throw new NullPointerException();
    int hash = hash(key);
    // 根据 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);
    // 插入新值到 槽 s 中
    return s.put(key, hash, value, false);
}

Segment 内部是由数组+链表组成的。

final V put(K key, int hash, V value, boolean onlyIfAbsent) {
    // 先获取该 segment 的独占锁
    // 每一个Segment进行put时，都会加锁
    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;
        // 数组该位置处的链表的表头
        HashEntry<K,V> first = entryAt(tab, index);
        for (HashEntry<K,V> e = first;;) {
            // 如果链头不为 null
            if (e != null) {
                K k;
                //如果在该链中找到相同的key，则用新值替换旧值，并退出循环
                if ((k = e.key) == key ||
                    (e.hash == hash && key.equals(k))) {
                    oldValue = e.value;
                    if (!onlyIfAbsent) {
                        e.value = value;
                        ++modCount;
                    }
                    break;
                }
                //如果没有和key相同的，一直遍历到链尾，链尾的next为null，进入到else
                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;
}

2.3 初始化Segment

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]; // use segment 0 as prototype
        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) { // recheck Segment[k] 是否被其它线程初始化了
            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;
}

2.4 get过程分析

比较简单，先找到 Segment 数组的位置，然后找到 HashEntry 数组的位置，最后顺着链表查找即可。

public V get(Object key) {
    Segment<K,V> s; // manually integrate access methods to reduce overhead
    HashEntry<K,V>[] tab;
    int h = hash(key);
    long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
    if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null &&
        (tab = s.table) != null) {
        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;
}

3 线程不安全

3.1 哈希碰撞

多个线程同时使用 put() 方法添加元素，若存在两个或多个 put() 的 key 发生了碰撞，那么有可能其中一个线程的数据被覆盖。

3.2 扩容

当数据要插入 HashMap 时，都会检查容量有没有超过设定的 thredhold，如果超过，则需要扩容。而多线程会导致扩容后的链表形成环形数据结构，一旦形成环形数据结构，Entry 的 next 的节点永远不为 null，就会在获取 Entry 时产生死循环。

例子可见文章《HashMap多线程死循环问题》。

不过要注意，其使用的 Java 版本既不是 7，也不是 8。在 Java7 中方法 addEntry() 添加节点到链表中是先扩容后再添加，而例子中的源码是：

void addEntry(int hash, K key, V value, int bucketIndex) {
    Entry<K,V> e = table[bucketIndex];
    // 先添加节点
    table[bucketIndex] = new Entry<K,V>(hash, key, value, e);
    
    // 然后扩容
    if (size++ >= threshold)
        resize(2 * table.length);
}

所以在 Java7 中此例子无效。而在 Java8 中，通过确保建新链与旧链的顺序是相同的，即可避免产生死循环。

4 HashMap遍历方式

import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
public class HashMapTest {
    private final static Map<Integer, Object> map = new HashMap<Integer, Object>(10000);
    private static final Object PRESENT = new Object();
    public static void main(String args[]) {
        long startTime;
        long endTime;
        long totalTime;
        for (int i = 0; i < 7500; i++) {
            map.put(i, PRESENT);
        }
        // 方法一
        startTime = System.nanoTime();
        Iterator iter1 = map.entrySet().iterator();
        while (iter1.hasNext()) {
            Map.Entry<Integer, Object> entry = (Map.Entry) iter1.next();
            Integer key = entry.getKey();
            Object val = entry.getValue();
        }
        endTime = System.nanoTime();
        totalTime = endTime - startTime;
        System.out.println("methor1 pays " + totalTime + " ms");
        
        // 方法二
        startTime = System.nanoTime();
        Iterator iter2 = map.keySet().iterator();
        while (iter2.hasNext()) {
            Object key = iter2.next();
            Object val = map.get(key);
        }
        endTime = System.nanoTime();
        totalTime = endTime - startTime;
        System.out.println("methor2 pays " + totalTime + " ms");
    }
}

运行结果：

深入剖析 Java 7 中的 HashMap 和 ConcurrentHashMap

5 性能对比

线程安全的使用 HashMap 有三种方式，分别为 Hashtable、SynchronizedMap()、ConcurrentHashMap。

Hashtable

使用 synchronized 来保证线程安全，几乎所有的 public 的方法都是 synchronized 的，而有些方法也是在内部通过 synchronized 代码块来实现。

synchronizedMap()

通过创建一个线程安全的 Map 对象，并把它作为一个封装的对象来返回。

ConcurrentHashMap

支持多线程对 Map 做读操作，并且不需要任何的 blocking 。这得益于 CHM 将 Map 分割成了不同的部分，在执行更新操作时只锁住一部分。根据默认的并发级别， Map 被分割成 16 个部分，并且由不同的锁控制。这意味着，同时最多可以有 16个写线程操作 Map 。试想一下，由只能一个线程进入变成同时可由 16 个写线程同时进入(读线程几乎不受限制)，性能的提升是显而易见的。但由于一些更新操作，如 put(), remove(), putAll(), clear()只锁住操作的部分，所以在检索操作不能保证返回的是最新的结果。

在迭代遍历 CHM 时， keySet 返回的 iterator 是弱一致和 fail-safe 的，可能不会返回某些最近的改变，并且在遍历过程中，如果已经遍历的数组上的内容变化了，不会抛出 ConcurrentModificationExceptoin 的异常。

什么时候使用 ConcurrentHashMap ？

CHM 适用于读者数量超过写者时，当写者数量大于等于读者时，CHM 的性能是低于 Hashtable 和 synchronizedMap 的。这是因为当锁住了整个 Map 时，读操作要等待对同一部分执行写操作的线程结束。

CHM 适用于做 cache ，在程序启动时初始化，之后可以被多个请求线程访问。

CHM 是Hashtable一个很好的替代，但要记住， CHM 的比 Hashrable 的同步性稍弱。

6 拓展：Java8 HashMap & ConcurrentHashMap

Java8 对 HashMap 和 ConcurrentHashMap 做了一些修改：

二者均利用了红黑树，所以其数据结构由数组 + 链表 + 红黑树组成。我们知道，链表上的数据需要一个一个比较下去才能找到我们需要的，时间复杂度取决于链表的长度，为 O(n)。为了降低这一部分的开销，在 Java8 中，当链表中的元素超过了 8 个以后，会将链表转换为红黑树，这个时候时间复杂度就降为了 O(logN)
Java8 中 ConcurrentHashMap 摒弃 Java7 中的 Segment 的概念，使用了另一种方式实现保证线程安全。

Linux公社的RSS地址： https://www.linuxidc.com/rssFeed.aspx

本文永久更新链接地址： https://www.linuxidc.com/Linux/2018-09/154133.htm

原文 https://www.linuxidc.com/Linux/2018-09/154133.htm

正文到此结束