HashMap与ConcurrentHashMap

2020-05-17

字数统计: 5.3k字 | 阅读时长≈ 22分

简介

在java的集合中，主要分为Collection和Map两大类，而在Map中，HashMap和ConcurrentHashMap无疑是非常重要并且常用的，下面主要介绍在JDK1.7和JDK1.8中的HashMap与ConcurrentHashMap

HashMap

JDK1.7中的HashMap

JDK1.7中的HashMap是由数组+链表的形式构成，没有红黑树。

JDK1.7的HashMap数据结构图：
strategy

1.7中HashMap比较核心的成员变量：
strategy
主要意思是：

初始化桶大小，因为底层是数组，所以这是数组的默认大小，为16
桶最大值，2^30
默认的负载因子，0.75
table：真正存放数据的数组
size：桶大小，可以在初始化时显式指定
负载因子，可以在初始化时显式指定

可以发现，1.7中的HashMap真正存放数据的是：

1	transient Entry<K,V>[] table = (Entry<K,V>[]) EMPTY_TABLE;

Entry<K,V>的默认实现：
strategy

Entry<K,V>是HashMap的一个内部类，主要有四个成员变量：

key：就是写入时的键
value：写入时的值
next：hashmap是数组+链表的形式实现，这个next就是为了链表的实现而存在
hash：存放的是当前key的hash值

put方法

源码：

public V put(K key, V value) {
    if (table == EMPTY_TABLE) {
        inflateTable(threshold);
    }
    if (key == null)
        return putForNullKey(value);
    int hash = hash(key);
    int i = indexFor(hash, table.length);
    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++;
    addEntry(hash, key, value, i);
    return null;
}

主要流程：

判断当前数组是否需要初始化
判断key是否是null值
计算key的hash值，并与数组长度进行与操作，得到索引值，定位出所在桶
如果对应桶是一个链表，那么遍历判断里面的hashcode、key是否与传入的key相同，如果有相同的，那么进行覆盖，并返回原来的值
如果对应桶是空的，说明此位置没有元素存入，那么生成一个Entry对象写入当前位置

addEntry方法：

//将 key-value 添加到索引为bucketIndex的桶处
void addEntry(int hash, K key, V value, int bucketIndex) {
    if ((size >= threshold) && (null != table[bucketIndex])) {
        resize(2 * table.length);
        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是桶的索引，所以e就是桶处已有链表的头结点
    Entry<K,V> e = table[bucketIndex];
    //new Entry的构造方法中，e是next元素，
    //所以这里就是将已有链表的头结点作为新节点的next，所以是头插法
    table[bucketIndex] = new Entry<>(hash, key, value, e);
    size++;
}

主要流程：

当调用addEntry时，会判断是否需要扩容：数组已有元素量超过门槛值，且此次插入索引处已有值
如果需要扩容，则进行两倍扩容，并且将当前key重新进行hash定位
在createEntry中，会将当前位置的桶传入到新建的桶中，如果桶有值就会在对应位置形成链表。

注意！！！addEntry方法中使用的是头插法，这为死循环埋下了伏笔

get方法

源码：

public V get(Object key) {
    if (key == null)
        return getForNullKey();
    Entry<K,V> entry = getEntry(key);
    return null == entry ? null : entry.getValue();
}
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;
}

主要流程：

首先也是根据key算出hashcode，然后定位到具体的桶中
判断该位置是否是链表
如果不是，就根据key和key的hashcode 是否相等来返回值
为链表，就需要遍历链表直到key和hashcode相等时才返回值
没有找到就返回null

JDK1.8中的HashMap

1.7中的HashMap存在一个很严重的问题，那就是当Hash冲突严重时，桶上的链表就会越来越长，这样在查询时的效率就会越来越低，时间复杂度接近O(N)
1.8 重点优化了这个查询效率。

1.8的HashMap结构图：
strategy

主要核心成员变量：

//数组初始化默认容量
static final int DEFAULT_INITIAL_CAPACITY = 1 << 4; // aka 16
//数组最大容量
static final int MAXIMUM_CAPACITY = 1 << 30;
//默认负载因子
static final float DEFAULT_LOAD_FACTOR = 0.75f;
//默认链表转红黑树的阈值大小
static final int TREEIFY_THRESHOLD = 8;
//存储元素的底层数组
transient Node<K,V>[] table;

//保存map中已有元素的集合，以Set形式保存
transient Set<Map.Entry<K,V>> entrySet;

//map中实际的元素数量
transient int size;

put方法

strategy

主要流程：

首先判断数组是否为空，如果空，就需要初始化
根据当前key的hashcode定位到具体的桶中并判断是否为空，为空表明没有Hash冲突，就直接在当前位置创建一个新桶即可。
如果当前桶有值（Hash冲突），那么就要比较当前桶key和hashcode与写入的key是否相等，相等就赋值给e，在第8步会统一赋值并返回
如果当前桶为红黑树，那就要按照红黑树的方式写入数据
如果是链表，就需要将当前的key、value封装成一个新节点写入到当前桶的后面（插尾法）
接着判断当前链表的大小是否大于预设的阈值，大于就需要转换为红黑树
如果在遍历过程中找到key相同时，直接退出遍历
如果e!=null，就相当于存在相同的key，那就需要将值覆盖
最后判断是否需要进行扩容

get方法

public V get(Object key) {
    Node<K,V> e;
    return (e = getNode(hash(key), key)) == null ? null : e.value;
}
//hash就是key的hash值
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;
}

主要流程：

首先将 key hash 之后取得所定位的桶。
如果桶为空则直接返回 null 。
否则判断桶的第一个位置(有可能是链表、红黑树)的 key 是否为查询的 key，是就直接返回 value。
如果第一个不匹配，则判断它的下一个是红黑树还是链表。
红黑树就按照树的查找方式返回值。
不然就按照链表的方式遍历匹配返回值。

HashMap的问题：
主要是并发场景下会出现线程安全问题：

对于jdk1.7而言，hashMap的put操作如果引发了resize()，就可能造成死循环。
对于jdk1.7和jdk1.8而言，hashMap线程不安全的共同特点就是没有针对多线程并发情况进行防护，比如没有对put、get操作做同步处理。

还有一个比较值得注意的是，hashMap底层数组的长度均为2的幂次方，是为了数组长度和key的hash值进行与操作时，让结果可以出现在数组的任意位置，避免某些位置始终得不到元素的存放，造成了资源浪费。

JDK1.7下的HashMap死循环问题

对于JDK1.7而言，内部使用的数据结构是数组+链表的方式，链表节点的插入方式为头插入。
JDK1.7中的Put操作逻辑：

判断数组是否需要初始化
判断key是否合法，是否能够插入

当允许插入元素时，会判断当前数组是否需要扩容。
关键就是扩容操作，扩容操作相关源码如下：

//扩容函数
void transfer(Entry[] newTable)
{
    Entry[] src = table;
    int newCapacity = newTable.length;
    //下面这段代码的意思是：
    //  从OldTable里摘一个元素出来，然后放到NewTable中
    for (int j = 0; j < src.length; j++) {
        Entry<K,V> e = src[j];
        if (e != null) {
            src[j] = null;
            do {
                Entry<K,V> next = e.next;
                int i = indexFor(e.hash, newCapacity);
                e.next = newTable[i];
                newTable[i] = e;
                e = next;
            } while (e != null);
        }
    }
}

//这就是最关键的transfer方法，当建立新数组之后，需要将原数组的元素转移过来
void transfer(Entry[] newTable)
{
    Entry[] src = table;
    int newCapacity = newTable.length;
    //下面这段代码的意思是：
    //  从OldTable里摘一个元素出来，然后放到NewTable中
    for (int j = 0; j < src.length; j++) {
        Entry<K,V> e = src[j];
        if (e != null) {
            src[j] = null;
            do {
                Entry<K,V> next = e.next;
                int i = indexFor(e.hash, newCapacity);
                e.next = newTable[i];
                newTable[i] = e;
                e = next;
            } while (e != null);
        }
    }
}

接下来以示例的方式说明死循环的产生：

现在假设HashMap内部元素有一部分是这样：
strategy

现在有两个线程，线程1和线程2需要添加元素，此时需要对HashMap分别进行扩容操作：

线程1执行到transfer方法的 Entry<K,V> next = e.next 时，时间片耗尽，停下了，此时对于线程1而言：
- e：Key3/Value:A e.next：Key9/Value:B
- next：Key9/Value:B
- HashMap的状态：
  原数组：
  
  新数组：
然后轮到了线程2开始执行transfer方法，线程2执行的比较顺利，一次执行结束，而比较巧合的是，Key3和Key9进行rehash计算之后，仍然在新数组的同一个索引处，并且由于从链表头部遍历节点，一个一个插入到新数组，并且对于链表是头插法，所以线程2的HashMap状态为：
新数组：

而线程2中的原数组已经被取代了

接着，再次轮到了线程1进行操作

此时线程1的 e是Key3/Value:A，next是Key9/Value:B；

执行do/while循环中接下来的语句：

int i = indexFor(e.hash, newCapacity);
e.next = newTable[i];
newTable[i] = e;
e = next;

然后线程1的数组状态就被改变了：
线程1的新数组状态：
此时 e就变成了Key9/Value:B，而此时的 Key9/Value:B的next已经被线程2修改为了Key:3/Value:A

线程1再次执行do/while语句

do {
    //e是 Key9/Value:B        next是 Key3/Value:A
    Entry<K,V> next = e.next;
    int i = indexFor(e.hash, newCapacity);
    //头插法
    e.next = newTable[i];
    newTable[i] = e;
    //e变成了Key3/Value:A
    e = next;
} while (e != null);

可以看到，当此次的do/while执行完之后，线程1新数组变成了：
strategy

注意！！！：此时e已经又变成了Key3/Value:A，所以下一次的do/while语句：

do {
    //e是 Key3/Value:A        next是 null
    Entry<K,V> next = e.next;
    int i = indexFor(e.hash, newCapacity);
    //注意！！！，这个时候的 newTable[i]是上一轮插入的Key9/Value:B
    e.next = newTable[i];
    newTable[i] = e;
    //e变成了Key9/Value:B
    e = next;
} while (e != null);

可以看到，执行完这次的do/while语句，就导致Key3/Value:A被重新插入到新数组，并且让Key3/Value:A的next变成了Key9/Value:B，而之前就已经知道了Key9/Value:B的next变成了Key3/Value:A，循环链就出现了，此时的新数组变成了：
strategy
从此之后，线程1就会陷入在do/while语句中死循环操作。

总结：
我们可以看到，JDK7中Put操作的死循环其实是因为扩容之后，链表的顺序被颠倒了，在JDK8中的HashMap，链表不再使用头插法，而是使用了尾插法，那么链表的各个元素之间的相对顺序不会被改变，所以JDK8的HashMap的扩容操作不会再出现死循环。
但是JDK8的HashMap仍然不是线程安全的，因为没有对put/get等操作进行同步保护。

ConcurrentHashMap

JDK1.7的ConcurrentHashMap

java7中的ConcurrentHashMap组成：Segment[]+HashEntry[]+链表
数据结构图如下：
strategy

通俗的来讲就是，ConcurrentHashMap中存在一个Segment[]数组，这个数组的每一个元素就是一个Segment类型的变量，而每一个Segment变量中都含有一个HashEntry[]数组，这个HashEntry数组才是真正用来存放<Key,Value>元素的数据结构。
可以理解为，一个ConcurrentHashMap中拥有很多个小的Map，每个Map就是一个Segment，底层存储数据的结构是HashEntry数组。

Segment是ConcurrentHashMap的一个内部类，主要组成：

static final class Segment<K,V> extends ReentrantLock implements Serializable {
       private static final long serialVersionUID = 2249069246763182397L;
       
       // 和 HashMap 中的 HashEntry 作用一样，真正存放数据的桶
       transient volatile HashEntry<K,V>[] table;
       
       transient int count;
       transient int modCount;
       transient int threshold;
       final float loadFactor;
       
}

HashEntry的组成：
strategy
可以看到，HashEntry与HashMap中的Entry十分相似，只不过HashEntry中的核心数据如Value、链表next都是volatile修饰的，保证了可见性。

到这里，我们大概知道了，ConcurrentHashMap采用了分段锁技术，其中分段采用的Segment继承于ReentrantLock，理论上ConcurrentHashMap支持Segment数组的length个线程的并发。每当一个线程占用某一个Segment的锁时，不会影响其他线程访问其他的Segment；

put方法

源码：

public V put(K key, V value) {
    Segment<K,V> s;
    if (value == null)
        throw new NullPointerException();
    int hash = hash(key);
    int j = (hash >>> segmentShift) & segmentMask;
    if ((s = (Segment<K,V>)UNSAFE.getObject          // nonvolatile; recheck
         (segments, (j << SSHIFT) + SBASE)) == null) //  in ensureSegment
        s = ensureSegment(j);
    return s.put(key, hash, value, false);
}

首先通过key的hash值定位到Segment，然后加锁，在对应的Segment中进行具体的Put操作

final V put(K key, int hash, V value, boolean onlyIfAbsent) {
    //对此Segment进行加锁，如果获取锁失败，就利用scanAndLockForPut()自旋获取锁
    HashEntry<K,V> node = tryLock() ? null :
        scanAndLockForPut(key, hash, value);
    V oldValue;
    try {
        //得到此Segment的底层存储数据的 HashEntry数组
        HashEntry<K,V>[] tab = table;
        int index = (tab.length - 1) & hash;
        //根据hash值找到对应的索引
        HashEntry<K,V> first = entryAt(tab, index);
        for (HashEntry<K,V> e = first;;) {
            //如果索引处已有值，就以链表的方式进行添加
            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;
                }
                e = e.next;
            }
            //找到了链表的末尾，链表前部分没有与key相同的已有元素
            else {
                //如果此索引已有值，就将key/value添加到链表末尾
                if (node != null)
                    node.setNext(first);
                else
                //此索引没有值，那么新建一个HashEntry
                    node = new HashEntry<K,V>(hash, key, value, first);
                int c = count + 1;
                //判断是否需要扩容
                if (c > threshold && tab.length < MAXIMUM_CAPACITY)
                    rehash(node);
                else
                    //不需要扩容/扩容结束了，将刚才新建的HashEntry存放在此索引处
                    setEntryAt(tab, index, node);
                ++modCount;
                count = c;
                oldValue = null;
                break;
            }
        }
    } finally {
        //put操作结束，释放持有的Segment对应的锁
        unlock();
    }
    return oldValue;
}

scanAndLockForPut()的源码：
strategy

尝试自旋获取锁
如果重试的次数达到了 MAX_SCAN_RETRIES，则改为阻塞锁获取，保证能获取成功

put方法的主要流程：

首先找到对应的Segment，接着将此Segment加锁，然后将当前 Segment 中的 table 通过 key 的 hashcode 定位到 HashEntry。
遍历该 HashEntry，如果不为空则判断传入的 key 和当前遍历的 key 是否相等，相等则覆盖旧的 value, 均不相等，则新建一个HashEntry，然后判断是否需要扩容，然后将新建的HashEntry添加到对应位置。
为空则需要新建一个 HashEntry 并加入到 Segment 中，同时会先判断是否需要扩容。
最后会解除在 1 中所获取当前 Segment 的锁。

get方法

源码：

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

主要流程：

根据Key通过Hash找到对应的Segment，然后再通过Hash定位到数组中的具体索引处
判断索引处是否已有值，有值或有链表，就遍历看是否存在Key，存在则返回，不存在返回null

由于 HashEntry 中的 value 属性是用 volatile 关键词修饰的，保证了内存可见性，所以每次获取时都是最新值。
ConcurrentHashMap 的 get 方法是非常高效的，因为整个过程都不需要加锁。

总结：
JDK7中的ConcurrentHashMap通过分段锁的方式保证了一定程度的并发量（主要是对Put方法加锁），通过底层数据结构HashEntry类对关键属性加上Volatile关键字保证了可见性。

JDK1.8中的ConcurrentHashMap

1.7 已经解决了并发问题，并且能支持 N 个 Segment 这么多次数的并发，但依然存在 HashMap 在 1.7 版本中的问题。

那就是查询遍历链表效率太低。

JDK1.8对ConcurrentHashMap的结构进行了一些调整：
strategy

JDK1.8抛弃了1.7中的Segment分段锁，而采用了 CAS+synchronized 来保证并发安全性
并将1.7中存放数据的HashEntry改为了Node，但作用是相同的：
strategy
其中val、next都用了volatile进行修饰，保证了可见性

在1.8的ConCurrentHashMap中，直接使用了 Node[]数组作为存储数据的底层数据结构，没有包装一层Segment。

put方法

源码：

final V putVal(K key, V value, boolean onlyIfAbsent) {
        if (key == null || value == null) throw new NullPointerException();
        //计算key的hash值
        int hash = spread(key.hashCode());
        int binCount = 0;
        for (Node<K,V>[] tab = table;;) {
            Node<K,V> f; int n, i, fh;
            //判断数组是否已经初始化了，如果没有，先初始化
            if (tab == null || (n = tab.length) == 0)
                tab = initTable();
            else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
                //f就是key对应hash值在数组中找到的索引处原有值
                //如果f==null，就说明这个索引处原来没有值
                if (casTabAt(tab, i, null,new Node<K,V>(hash, key, value, null))){
                    //利用CAS操作将key/value存入到此索引，结束put操作
                    break;  
                }                
            }
            //如果数组正在扩容，那么此线程就去帮助数组进行扩容
            else if ((fh = f.hash) == MOVED)
                tab = helpTransfer(tab, f);
            else {
            //如果此索引处已有值，且数组没有在扩容
                V oldVal = null;
            //对此索引处已有值的头节点加锁，也就是对此索引加锁
                synchronized (f) {
                    if (tabAt(tab, i) == f) {
                        //如果是链表形式，尾插法添加元素
                        if (fh >= 0) {
                            binCount = 1;
                            for (Node<K,V> e = f;; ++binCount) {
                                K ek;
                                if (e.hash == hash &&
                                    ((ek = e.key) == key ||
                                     (ek != null && key.equals(ek)))) {
                                    oldVal = e.val;
                                    if (!onlyIfAbsent)
                                        e.val = value;
                                    break;
                                }
                                Node<K,V> pred = e;
                                if ((e = e.next) == null) {
                                    pred.next = new Node<K,V>(hash, key,
                                                              value, null);
                                    break;
                                }
                            }
                        }
                        //如果是红黑树形式，以红黑树的方式添加新key/value
                        else if (f instanceof TreeBin) {
                            Node<K,V> p;
                            binCount = 2;
                            if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
                                                           value)) != null) {
                                oldVal = p.val;
                                if (!onlyIfAbsent)
                                    p.val = value;
                            }
                        }
                    }
                }
                //判断是否需要进行链表-红黑树转换
                if (binCount != 0) {
                    if (binCount >= TREEIFY_THRESHOLD)
                        treeifyBin(tab, i);
                    if (oldVal != null)
                        return oldVal;
                    break;
                }
            }
        }
        addCount(1L, binCount);
        return null;
    }

主要流程：

根据 key 计算出 hashcode 。
判断是否需要进行初始化。
f 即为当前 key 定位出的 Node，如果为空表示当前位置可以写入数据，利用 CAS 尝试写入，失败则自旋保证成功。
果当前位置的 hashcode == MOVED == -1,则需要进行扩容。
如果都不满足，则利用 synchronized 锁写入数据。
如果数量大于 TREEIFY_THRESHOLD 则要转换为红黑树。

值得注意的是，1.8的ConcurrentHashMap是对数组的单个索引进行加锁，锁粒度更细了，能够支持更大容量的并发操作。

get方法

源码：

public V get(Object key) {
        Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
        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)
                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;
    }

主要流程：

根据计算出来的 hashcode 寻址，如果就在桶上那么直接返回值。
如果是红黑树那就按照树的方式获取值。
就不满足那就按照链表的方式遍历获取值。

总结：
JDK1.8的ConcurrentHashMap的改进在于锁的粒度更细了，针对于数组的单个索引进行加锁，并且引入了红黑树结构，查询效率更高。

参考

本文作者： xczll
本文链接： https://xczllgit.github.io/2020/05/17/javaSourceCode/2020-05-17-hashMap-ConcurrentHashMap/
版权声明： 本博客所有文章除特别声明外，均采用 MIT 许可协议。转载请注明出处！