javaSE 面试题02:HashMap


1. 一个类的构造方法的作用是什么?构造方法有哪些特性?

构造函数用于完成对象的初始化工作。

如果一个类没有声明构造函数,编译器会 为类生成默认的、不带参数的构造方法。

构造函数的特性:

1,名字与类名相同;

2,没有返回值,但不能用void声明构造函数

3,生成类的对象时自动执行,无需调用

2. 静态方法和实例方法有何不同?

两个方面:

  1. 在外部调用静态方法时,可以使用”类名.方法名”的方式,也可以使用”对象名.方法名”的方式。而实例方法只有后面这种方式。也就是说,调用静态方法可以无需创建对象。
  2. 静态方法在访问本类的成员时,只允许访问静态成员(即静态成员变量和静态方法),而不允许访问实例成员变量和实例方法;实例方法则无此限制

3. String对象是不是真的是不可变的?

  1. String不可变,是说实例对象中的数组不可变。
  2. 但是,事实上是可以通过反射技术改变数组的

*4. 什么是反射?

什么是反射?

在运行状态中,对于任意一个类,能够获取这个类的所有属性和方法,并且可以实例化。

在运行状态中,对于任意一个对象,能够获取这个实例的类的所有属性和方法,能够调用对象的任意一个方法和属性;

这种动态获取信息、以及动态调用对象方法的功能,称为java语言的反射机制。

静态编译和动态编译

静态编译:代码写死,在编译时类型确定,绑定对象

动态编译:运行时确定类型,绑定对象

反射机制优缺点

优点: 灵活

缺点: 影响性能

5. List,Set,Map三者的区别:

List: 有序,可重复

Set: 不可重复,无序

Map: Key-Value,不可重复

*6. ArrayList、LinkedList、Vector三者的区别:

​ 继承关系:

img

ArrayList: 底层使用数组,随机读取速度快(O(1)),但是插入删除需要移动数据(O(N))

LinkedList: 底层使用双向链表,随机读取速度慢(O(N)),但是插入删除不需要移动数据

Vector: 底层使用数组,Vector是线程安全的,但是性能比ArrayList要低,内部用的同步;ArrayList和LinkedList都不是线程安全的。

*7. HashMap、HashSet、HashTable三者的区别:

继承关系:

img

HashMap 和 HashSet 是 Java Collection 的两个重要成员,其中 HashMap 是 Map 接口的常用实现类,HashSet 是 Set 接口的常用实现类。虽然 HashMap 和 HashSet 实现的接口规范不同,但它们底层的 Hash 存储机制完全一样,甚至 HashSet 本身就采用 HashMap 来实现的。

HashMap: Key-Value方式,底层是哈希表,Key的判等是根据HashCode()和equals()的返回决定,相同的Key,后面的写操作会覆盖以前的数据。

  HashSet: 使用实例本身计算HashCode当做key,本质也是HashMap,JDK1.8前底层是哈希数组+链表,JDK1.8开始是哈希数组+链表(个数大于8个)+红黑树(个数小于8个)

  HashTable: HashTable和HashMap很接近,区别仅在于HashTable是线程安全的,而前两者不是;另外,HashTable不允许Key是null,但是HashMap允许。

*8. HashMap、ConcurrentHashMap的区别

HashMap: HashMap不是线程安全的;允许Key是null

ConcurrentHashMap: ConcurrentHashMap是线程安全的,与HashTable相比,HashTable使用的是synchronized的锁,而ConcurrentHashMap则将自己的Key空间进行了分段(Segment)加锁(在 1.7 以前使用的是分段锁机制实现的线程安全的。但是在 1.8 以后使用 synchronized 关键字实现的线程安全,并且在JDK1.8开始使用CAS),所以ConcurrentHashMap比HashTable具有更高的并发性能;不允许Key是null。

*9.comparable 和 comparator的区别?

comparable: 接口,需要类Override compareTo方法,一个类只能有一种compareTo方法。Comparable是排序接口。若一个类实现了Comparable接口,就意味着该类支持排序。实际上所有实现了 Comparable 接口的 Java 核心类的结果都和 equlas 方法保持一致。实现了 Comparable 接口的 List 或者数组可以使用 Collections.sort() 或者 Arrays.sort() 方法进行排序。

int compare(T o1, T o2) 是“比较o1和o2的大小”。返回“负数”,意味着“o1比o2小”;返回“零”,意味着“o1等于o2”;返回“正数”,意味着“o1大于o2”。

Comparable 可以让实现它的类的对象进行比较,具体的比较规则是按照 compareTo 方法中的规则进行。这种顺序称为 自然顺序

compareTo 方法的返回值有三种情况:

  • e1.compareTo(e2) > 0 即 e1 > e2
  • e1.compareTo(e2) = 0 即 e1 = e2
  • e1.compareTo(e2) < 0 即 e1 < e2

1.由于 null 不是一个类,也不是一个对象,因此在重写 compareTo 方法时应该注意 e.compareTo(null) 的情况,即使 e.equals(null) 返回 false,compareTo 方法也应该主动抛出一个空指针异常 NullPointerException。

2.Comparable 实现类重写 compareTo 方法时一般要求 e1.compareTo(e2) == 0 的结果要和 e1.equals(e2) 一致。这样将来使用 SortedSet 等根据类的自然排序进行排序的集合容器时可以保证保存的数据的顺序和想象中一致。举个例子,如果你往一个 SortedSet 中先后添加两个对象 a 和 b,a b 满足 (!a.equals(b) && a.compareTo(b) == 0),同时也没有另外指定个 Comparator,那当你添加完 a 再添加 b 时会添加失败返回 false, SortedSet 的 size 也不会增加,因为在 SortedSet 看来它们是相同的,而 SortedSet 中是不允许重复的。

comparator: 接口,实现该接口的比较类是独立于被比较的类的,比较类需要Override compare方法,一个类可以有多个不同的比较类。

从上面内容可知使用自然排序需要类实现 Comparable,并且在内部重写 comparaTo 方法。而 Comparator 则是在外部制定排序规则,然后作为排序策略参数传递给某些类,比如 Collections.sort(), Arrays.sort(), 或者一些内部有序的集合(比如 SortedSet,SortedMap 等)。

若一个类要实现Comparator接口:它一定要实现compare(T o1, T o2) 函数,但可以不实现 equals(Object obj) 函数。

Java 中的两种排序方式:

  1. Comparable 自然排序。(实体类实现)
  2. Comparator 是定制排序。(无法修改实体类时,直接在调用方创建)

10. 如何对Object的list排序?

​ 对objects数组进行排序,我们可以用Arrays.sort()方法。

​ 对objects的集合进行排序,需要使用Collections.sort()方法。

*11. 简述HashMap的工作原理以及分析源码

关键思路:1.简单说一下, JDK1.8前底层是哈希数组+链表,JDK1.8开始是哈希数组+链表(个数小于8个)+红黑树(个数大于8个)  2.介绍哈希表存储位置的计算(哈希函数)3. Hash冲突  4.哈希表数据结构以及相关操作时间复杂度 5.影响因子  

简介

HashMap主要用来存放键值对,它基于哈希表的Map接口实现,是常用的Java集合之一。与HashTable主要区别为不支持同步和允许null作为key和value,所以如果你想要保证线程安全,可以使用ConcurrentHashMap代替而不是线程安全的HashTable,因为HashTable基本已经被淘汰。

内部结构分析

JDK1.8之前:

JDK1.8之前HashMap底层是数组和链表结合在一起使用也就是链表散列。HashMap通过key的hashCode来计算hash值,当hashCode相同时,通过“拉链法”解决冲突。

所谓“拉链法”就是:将链表和数组相结合。也就是说创建一个链表数组,数组中每一格就是一个链表。若遇到哈希冲突,则将冲突的值加到链表中即可。

jdk1.8之前的内部结构

简单来说,JDK1.8之前HashMap由数组+链表组成的,数组是HashMap的主体,链表则是主要为了解决哈希冲突而存在的,如果定位到的数组位置不含链表(当前entry的next指向null),那么对于查找,添加等操作很快,仅需一次寻址即可;如果定位到的数组包含链表,对于添加操作,其时间复杂度依然为O(1),因为最新的Entry会插入链表头部,急需要简单改变引用链即可,而对于查找操作来讲,此时就需要遍历链表,然后通过key对象的equals方法逐一比对查找.

JDK1.8之后:

相比于之前的版本,jdk1.8在解决哈希冲突时有了较大的变化,当链表长度大于阈值(默认为8)时,将链表转化为红黑树(树化),以减少搜索时间。转化为红黑树后,如果后续操作又导致链表长度小于6时,此时又会转化为链表(链化)。

JDK1.8之后的内部结构

什么是哈希表?

在讨论哈希表之前,我们先大概了解下其他数据结构在新增,查找等基础操作执行性能

  数组:采用一段连续的存储单元来存储数据。对于指定下标的查找,时间复杂度为O(1);通过给定值进行查找,需要遍历数组,逐一比对给定关键字和数组元素,时间复杂度为O(n),当然,对于有序数组,则可采用二分查找,插值查找,斐波那契查找等方式,可将查找复杂度提高为O(logn);对于一般的插入删除操作,涉及到数组元素的移动,其平均复杂度也为O(n)

  线性链表:对于链表的新增,删除等操作(在找到指定操作位置后),仅需处理结点间的引用即可,时间复杂度为O(1),而查找操作需要遍历链表逐一进行比对,复杂度为O(n)

  二叉树:对一棵相对平衡的有序二叉树,对其进行插入,查找,删除等操作,平均复杂度均为O(logn)。

  哈希表:相比上述几种数据结构,在哈希表中进行添加,删除,查找等操作,性能十分之高,不考虑哈希冲突的情况下,仅需一次定位即可完成,时间复杂度为O(1),接下来我们就来看看哈希表是如何实现达到惊艳的常数阶O(1)的。

  我们知道,数据结构的物理存储结构只有两种:顺序存储结构链式存储结构(像栈,队列,树,图等是从逻辑结构去抽象的,映射到内存中,也这两种物理组织形式),而在上面我们提到过,在数组中根据下标查找某个元素,一次定位就可以达到,哈希表利用了这种特性,哈希表的主干就是数组

  比如我们要新增或查找某个元素,我们通过把当前元素的关键字 通过某个函数映射到数组中的某个位置,通过数组下标一次定位就可完成操作。

        存储位置 = f(关键字)

  其中,这个函数f一般称为哈希函数,这个函数的设计好坏会直接影响到哈希表的优劣。举个例子,比如我们要在哈希表中执行插入操作:

  img

  查找操作同理,先通过哈希函数计算出实际存储地址,然后从数组中对应地址取出即可。

  哈希冲突

  然而万事无完美,如果两个不同的元素,通过哈希函数得出的实际存储地址相同怎么办?也就是说,当我们对某个元素进行哈希运算,得到一个存储地址,然后要进行插入的时候,发现已经被其他元素占用了,其实这就是所谓的哈希冲突,也叫哈希碰撞。前面我们提到过,哈希函数的设计至关重要,好的哈希函数会尽可能地保证 计算简单散列地址分布均匀,但是,我们需要清楚的是,数组是一块连续的固定长度的内存空间,再好的哈希函数也不能保证得到的存储地址绝对不发生冲突。那么哈希冲突如何解决呢?哈希冲突的解决方案有多种:开放定址法(发生冲突,继续寻找下一块未被占用的存储地址),再散列函数法,链地址法,而HashMap即是采用了链地址法,又名拉链法,也就是数组+链表的方式。

类的属性:

public class HashMap<K,V> extends AbstractMap<K,V> implements Map<K,V>, Cloneable, Serializable {
    // 序列号
    private static final long serialVersionUID = 362498820763181265L;    
    // 默认的初始容量是16
    static final int DEFAULT_INITIAL_CAPACITY = 1 << 4;   
    // 最大容量
    static final int MAXIMUM_CAPACITY = 1 << 30; 
    // 默认的填充因子
    static final float DEFAULT_LOAD_FACTOR = 0.75f;
    // 当桶(bucket)上的结点数大于这个值时会转成红黑树
    static final int TREEIFY_THRESHOLD = 8; 
    // 当桶(bucket)上的结点数小于这个值时树转链表
    static final int UNTREEIFY_THRESHOLD = 6;
    // 桶中结构转化为红黑树对应的table的最小大小
    static final int MIN_TREEIFY_CAPACITY = 64;
    // 存储元素的数组,总是2的幂次倍
    transient Node<k,v>[] table; 
    // 存放具体元素的集
    transient Set<map.entry<k,v>> entrySet;
    // 存放元素的个数,注意这个不等于数组的长度。
    transient int size;
    // 每次扩容和更改map结构的计数器
    transient int modCount;   
    // 临界值 当实际大小(容量*填充因子)超过临界值时,会进行扩容
    int threshold;
    // 填充因子
    final float loadFactor;
}

HashMap实现原理

HashMap的主干是一个Entry数组。Entry是HashMap的基本组成单元,每一个Entry包含一个key-value键值对。

//HashMap的主干数组,可以看到就是一个Entry数组,初始值为空数组{},主干数组的长度一定是2的次幂,至于为什么这么做,后面会有详细分析。
transient Entry<K,V>[] table = (Entry<K,V>[]) EMPTY_TABLE;

Entry是HashMap中的一个静态内部类。代码如下

static class Entry<K,V> implements Map.Entry<K,V> {
    final K key;
    V value;
    Entry<K,V> next;//存储指向下一个Entry的引用,单链表结构
    int hash;//对key的hashcode值进行hash运算后得到的值,存储在Entry,避免重复计算
    /**
         * Creates new entry.
         */
    Entry(int h, K k, V v, Entry<K,V> n) {
        value = v;
        next = n;
        key = k;
        hash = h;
    } 

所以,HashMap的整体结构如下

img  

  简单来说,HashMap由数组+链表组成的,数组是HashMap的主体,链表则是主要为了解决哈希冲突而存在的,如果定位到的数组位置不含链表(当前entry的next指向null),那么对于查找,添加等操作很快,仅需一次寻址即可;如果定位到的数组包含链表,对于添加操作,其时间复杂度为O(n),首先遍历链表,存在即覆盖,否则新增;对于查找操作来讲,仍需遍历链表,然后通过key对象的equals方法逐一比对查找。所以,性能考虑,HashMap中的链表出现越少,性能才会越好。

其他几个重要字段

//实际存储的key-value键值对的个数
transient int size;
//阈值,当table == {}时,该值为初始容量(初始容量默认为16);当table被填充了,也就是为table分配内存空间后,threshold一般为 capacity*loadFactory。HashMap在进行扩容时需要参考threshold,后面会详细谈到
int threshold;
//负载因子,代表了table的填充度有多少,默认是0.75
final float loadFactor;
//用于快速失败,由于HashMap非线程安全,在对HashMap进行迭代时,如果期间其他线程的参与导致HashMap的结构发生变化了(比如put,remove等操作),需要抛出异常ConcurrentModificationException
transient int modCount;

HashMap有4个构造器,其他构造器如果用户没有传入initialCapacity 和loadFactor这两个参数,会使用默认值

initialCapacity默认为16,loadFactory默认为0.75

我们看下其中一个

public HashMap(int initialCapacity, float loadFactor) {     //此处对传入的初始容量进行校验,最大不能超过MAXIMUM_CAPACITY = 1<<30(230)
        if (initialCapacity < 0)
            throw new IllegalArgumentException("Illegal initial capacity: " +
                                               initialCapacity);
        if (initialCapacity > MAXIMUM_CAPACITY)
            initialCapacity = MAXIMUM_CAPACITY;
        if (loadFactor <= 0 || Float.isNaN(loadFactor))
            throw new IllegalArgumentException("Illegal load factor: " +
                                               loadFactor);

        this.loadFactor = loadFactor;
        threshold = initialCapacity;     
        init();//init方法在HashMap中没有实际实现,不过在其子类如 linkedHashMap中就会有对应实现
    }

  从上面这段代码我们可以看出,在常规构造器中,没有为数组table分配内存空间(有一个入参为指定Map的构造器例外),而是在执行put操作的时候才真正构建table数组

  OK,接下来我们来看看put操作的实现吧

    public V put(K key, V value) {
        //如果table数组为空数组{},进行数组填充(为table分配实际内存空间),入参为threshold,此时threshold为initialCapacity 默认是1<<4(24=16)
        if (table == EMPTY_TABLE) {
            inflateTable(threshold);
        }
       //如果key为null,存储位置为table[0]或table[0]的冲突链上
        if (key == null)
            return putForNullKey(value);
        int hash = hash(key);//对key的hashcode进一步计算,确保散列均匀
        int i = indexFor(hash, table.length);//获取在table中的实际位置
        for (Entry<K,V> e = table[i]; e != null; e = e.next) {
        //如果该对应数据已存在,执行覆盖操作。用新value替换旧value,并返回旧value
            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++;//保证并发访问时,若HashMap内部结构发生变化,快速响应失败
        addEntry(hash, key, value, i);//新增一个entry
        return null;
    }    

先来看看inflateTable这个方法

private void inflateTable(int toSize) {
        int capacity = roundUpToPowerOf2(toSize);//capacity一定是2的次幂
        threshold = (int) Math.min(capacity * loadFactor, MAXIMUM_CAPACITY + 1);//此处为threshold赋值,取capacity*loadFactor和MAXIMUM_CAPACITY+1的最小值,capaticy一定不会超过MAXIMUM_CAPACITY,除非loadFactor大于1
        table = new Entry[capacity];
        initHashSeedAsNeeded(capacity);
    }

  inflateTable这个方法用于为主干数组table在内存中分配存储空间,通过roundUpToPowerOf2(toSize)可以确保capacity为大于或等于toSize的最接近toSize的二次幂,比如toSize=13,则capacity=16;to_size=16,capacity=16;to_size=17,capacity=32.

 private static int roundUpToPowerOf2(int number) {
        // assert number >= 0 : "number must be non-negative";
        return number >= MAXIMUM_CAPACITY
                ? MAXIMUM_CAPACITY
                : (number > 1) ? Integer.highestOneBit((number - 1) << 1) : 1;
    }

roundUpToPowerOf2中的这段处理使得数组长度一定为2的次幂,Integer.highestOneBit是用来获取最左边的bit(其他bit位为0)所代表的数值.

hash函数

//这是一个神奇的函数,用了很多的异或,移位等运算,对key的hashcode进一步进行计算以及二进制位的调整等来保证最终获取的存储位置尽量分布均匀
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);
    }

以上hash函数计算出的值,通过indexFor进一步处理来获取实际的存储位置

  /**
     * 返回数组下标
     */
    static int indexFor(int h, int length) {
        return h & (length-1);
    }

h&(length-1)保证获取的index一定在数组范围内,举个例子,默认容量16,length-1=15,h=18,转换成二进制计算为

        1  0  0  1  0
    &   0  1  1  1  1
    __________________
        0  0  0  1  0    = 2

  最终计算出的index=2。有些版本的对于此处的计算会使用 取模运算,也能保证index一定在数组范围内,不过位运算对计算机来说,性能更高一些(HashMap中有大量位运算)

所以最终存储位置的确定流程是这样的:

img

再来看看addEntry的实现:

void addEntry(int hash, K key, V value, int bucketIndex) {
        if ((size >= threshold) && (null != table[bucketIndex])) {
            resize(2 * table.length);//当size超过临界阈值threshold,并且即将发生哈希冲突时进行扩容
            hash = (null != key) ? hash(key) : 0;
            bucketIndex = indexFor(hash, table.length);
        }

        createEntry(hash, key, value, bucketIndex);
    }

  通过以上代码能够得知,当发生哈希冲突并且size大于阈值的时候,需要进行数组扩容,扩容时,需要新建一个长度为之前数组2倍的新的数组,然后将当前的Entry数组中的元素全部传输过去,扩容后的新数组长度为之前的2倍,所以扩容相对来说是个耗资源的操作。

(1)loadFactor加载因子

loadFactor加载因子是控制数组存放数据的疏密程度,loadFactor越趋近于1,那么 数组中存放的数据(entry)也就越多,也就越密,也就是会让链表的长度增加,load Factor越小,也就是趋近于0,

loadFactor太大导致查找元素效率低,太小导致数组的利用率低,存放的数据会很分散。loadFactor的默认值为0.75f是官方给出的一个比较好的临界值。  

(2)threshold

threshold = capacity * loadFactor当Size>=threshold的时候,那么就要考虑对数组的扩增了,也就是说,这个的意思就是 衡量数组是否需要扩增的一个标准

为什么HashMap的数组长度一定是2的次幂?

我们来继续看上面提到的resize方法

 void resize(int newCapacity) {
        Entry[] oldTable = table;
        int oldCapacity = oldTable.length;
        if (oldCapacity == MAXIMUM_CAPACITY) {
            threshold = Integer.MAX_VALUE;
            return;
        }

        Entry[] newTable = new Entry[newCapacity];
        transfer(newTable, initHashSeedAsNeeded(newCapacity));
        table = newTable;
        threshold = (int)Math.min(newCapacity * loadFactor, MAXIMUM_CAPACITY + 1);
    }

如果数组进行扩容,数组长度发生变化,而存储位置 index = h&(length-1),index也可能会发生变化,需要重新计算index,我们先来看看transfer这个方法

void transfer(Entry[] newTable, boolean rehash) {
    int newCapacity = newTable.length;
    //for循环中的代码,逐个遍历链表,重新计算索引位置,将老数组数据复制到新数组中去(数组不存储实际数据,所以仅仅是拷贝引用而已)
    for (Entry<K,V> e : table) {
        while(null != e) {
            Entry<K,V> next = e.next;
            if (rehash) {
                e.hash = null == e.key ? 0 : hash(e.key);
            }
            int i = indexFor(e.hash, newCapacity);
            //将当前entry的next链指向新的索引位置,newTable[i]有可能为空,有可能也是个entry链,如果是entry链,直接在链表头部插入。
            e.next = newTable[i];
            newTable[i] = e;
            e = next;
        }
    }
}

  这个方法将老数组中的数据逐个链表地遍历,扔到新的扩容后的数组中,我们的数组索引位置的计算是通过 对key值的hashcode进行hash扰乱运算后,再通过和 length-1进行位运算得到最终数组索引位置。

  hashMap的数组长度一定保持2的次幂,比如16的二进制表示为 10000,那么length-1就是15,二进制为01111,同理扩容后的数组长度为32,二进制表示为100000,length-1为31,二进制表示为011111。从下图可以我们也能看到这样会保证低位全为1,而扩容后只有一位差异,也就是多出了最左位的1,这样在通过 h&(length-1)的时候,只要h对应的最左边的那一个差异位为0,就能保证得到的新的数组索引和老数组索引一致(大大减少了之前已经散列良好的老数组的数据位置重新调换),个人理解。

  img

还有,数组长度保持2的次幂,length-1的低位都为1,会使得获得的数组索引index更加均匀,比如:

img

  我们看到,上面的&运算,高位是不会对结果产生影响的(hash函数采用各种位运算可能也是为了使得低位更加散列),我们只关注低位bit,如果低位全部为1,那么对于h低位部分来说,任何一位的变化都会对结果产生影响,也就是说,要得到index=21这个存储位置,h的低位只有这一种组合。这也是数组长度设计为必须为2的次幂的原因。

img

  如果不是2的次幂,也就是低位不是全为1此时,要使得index=21,h的低位部分不再具有唯一性了,哈希冲突的几率会变的更大,同时,index对应的这个bit位无论如何不会等于1了,而对应的那些数组位置也就被白白浪费了。

get方法

 public V get(Object key) {     //如果key为null,则直接去table[0]处去检索即可。
        if (key == null)
            return getForNullKey();
        Entry<K,V> entry = getEntry(key);
        return null == entry ? null : entry.getValue();
 }

get方法通过key值返回对应value,如果key为null,直接去table[0]处检索。我们再看一下getEntry这个方法

final Entry<K,V> getEntry(Object key) {
            
        if (size == 0) {
            return null;
        }
        //通过key的hashcode值计算hash值
        int hash = (key == null) ? 0 : hash(key);
        //indexFor (hash&length-1) 获取最终数组索引,然后遍历链表,通过equals方法比对找出对应记录
        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;
    }    

  可以看出,get方法的实现相对简单,key(hashcode)–>hash–>indexFor–>最终索引位置,找到对应位置table[i],再查看是否有链表,遍历链表,通过key的equals方法比对查找对应的记录。要注意的是,有人觉得上面在定位到数组位置之后然后遍历链表的时候,e.hash == hash这个判断没必要,仅通过equals判断就可以。其实不然,试想一下,如果传入的key对象重写了equals方法却没有重写hashCode,而恰巧此对象定位到这个数组位置,如果仅仅用equals判断可能是相等的,但其hashCode和当前对象不一致,这种情况,根据Object的hashCode的约定,不能返回当前对象,而应该返回null,后面的例子会做出进一步解释。

Node节点类源码:

// 继承自 Map.Entry<K,V>
static class Node<K,V> implements Map.Entry<K,V> {
       final int hash;// 哈希值,存放元素到hashmap中时用来与其他元素hash值比较
       final K key;//键
       V value;//值
       // 指向下一个节点
       Node<K,V> next;
       Node(int hash, K key, V value, Node<K,V> next) {
            this.hash = hash;
            this.key = key;
            this.value = value;
            this.next = next;
        }
        public final K getKey()        { return key; }
        public final V getValue()      { return value; }
        public final String toString() { return key + "=" + value; }
        // 重写hashCode()方法
        public final int hashCode() {
            return Objects.hashCode(key) ^ Objects.hashCode(value);
        }

        public final V setValue(V newValue) {
            V oldValue = value;
            value = newValue;
            return oldValue;
        }
        // 重写 equals() 方法
        public final boolean equals(Object o) {
            if (o == this)
                return true;
            if (o instanceof Map.Entry) {
                Map.Entry<?,?> e = (Map.Entry<?,?>)o;
                if (Objects.equals(key, e.getKey()) &&
                    Objects.equals(value, e.getValue()))
                    return true;
            }
            return false;
        }
}

树节点类源码:

static final class TreeNode<K,V> extends LinkedHashMap.Entry<K,V> {
        TreeNode<K,V> parent;  // 父
        TreeNode<K,V> left;    // 左
        TreeNode<K,V> right;   // 右
        TreeNode<K,V> prev;    // needed to unlink next upon deletion
        boolean red;           // 判断颜色
        TreeNode(int hash, K key, V val, Node<K,V> next) {
            super(hash, key, val, next);
        }
        // 返回根节点
        final TreeNode<K,V> root() {
            for (TreeNode<K,V> r = this, p;;) {
                if ((p = r.parent) == null)
                    return r;
                r = p;
       }
复制代码

LinkedList源码分析

构造方法

四个构造方法

    // 默认构造函数。
    public More ...HashMap() {
        this.loadFactor = DEFAULT_LOAD_FACTOR; // all   other fields defaulted
     }
     
     // 包含另一个“Map”的构造函数
     public More ...HashMap(Map<? extends K, ? extends V> m) {
         this.loadFactor = DEFAULT_LOAD_FACTOR;
         putMapEntries(m, false);//下面会分析到这个方法
     }
     
     // 指定“容量大小”的构造函数
     public More ...HashMap(int initialCapacity) {
         this(initialCapacity, DEFAULT_LOAD_FACTOR);
     }
     
     // 指定“容量大小”和“加载因子”的构造函数
     public More ...HashMap(int initialCapacity, float loadFactor) {
         if (initialCapacity < 0)
             throw new IllegalArgumentException("Illegal initial capacity: " + initialCapacity);
         if (initialCapacity > MAXIMUM_CAPACITY)
             initialCapacity = MAXIMUM_CAPACITY;
         if (loadFactor <= 0 || Float.isNaN(loadFactor))
             throw new IllegalArgumentException("Illegal load factor: " + loadFactor);
         this.loadFactor = loadFactor;
         this.threshold = tableSizeFor(initialCapacity);
     }

putMapEntries方法:

final void putMapEntries(Map<? extends K, ? extends V> m, boolean evict) {
    int s = m.size();
    if (s > 0) {
        // 判断table是否已经初始化
        if (table == null) { // pre-size
            // 未初始化,s为m的实际元素个数
            float ft = ((float)s / loadFactor) + 1.0F;
            int t = ((ft < (float)MAXIMUM_CAPACITY) ?
                    (int)ft : MAXIMUM_CAPACITY);
            // 计算得到的t大于阈值,则初始化阈值
            if (t > threshold)
                threshold = tableSizeFor(t);
        }
        // 已初始化,并且m元素个数大于阈值,进行扩容处理
        else if (s > threshold)
            resize();
        // 将m中的所有元素添加至HashMap中
        for (Map.Entry<? extends K, ? extends V> e : m.entrySet()) {
            K key = e.getKey();
            V value = e.getValue();
            putVal(hash(key), key, value, false, evict);
        }
    }
}

put方法

HashMap只提供了put用于添加元素,putVal方法只是给put方法调用的一个方法,并没有提供给用户使用。

public V put(K key, V value) {
    return putVal(hash(key), key, value, false, true);
}

final V putVal(int hash, K key, V value, boolean onlyIfAbsent,
                   boolean evict) {
    Node<K,V>[] tab; Node<K,V> p; int n, i;
    // table未初始化或者长度为0,进行扩容
    if ((tab = table) == null || (n = tab.length) == 0)
        n = (tab = resize()).length;
    // (n - 1) & hash 确定元素存放在哪个桶中,桶为空,新生成结点放入桶中(此时,这个结点是放在数组中)
    if ((p = tab[i = (n - 1) & hash]) == null)
        tab[i] = newNode(hash, key, value, null);
    // 桶中已经存在元素
    else {
        Node<K,V> e; K k;
        // 比较桶中第一个元素(数组中的结点)的hash值相等,key相等
        if (p.hash == hash &&
            ((k = p.key) == key || (key != null && key.equals(k))))
                // 将第一个元素赋值给e,用e来记录
                e = p;
        // hash值不相等,即key不相等;为红黑树结点
        else if (p instanceof TreeNode)
            // 放入树中
            e = ((TreeNode<K,V>)p).putTreeVal(this, tab, hash, key, value);
        // 为链表结点
        else {
            // 在链表最末插入结点
            for (int binCount = 0; ; ++binCount) {
                // 到达链表的尾部
                if ((e = p.next) == null) {
                    // 在尾部插入新结点
                    p.next = newNode(hash, key, value, null);
                    // 结点数量达到阈值,转化为红黑树
                    if (binCount >= TREEIFY_THRESHOLD - 1) // -1 for 1st
                        treeifyBin(tab, hash);
                    // 跳出循环
                    break;
                }
                // 判断链表中结点的key值与插入的元素的key值是否相等
                if (e.hash == hash &&
                    ((k = e.key) == key || (key != null && key.equals(k))))
                    // 相等,跳出循环
                    break;
                // 用于遍历桶中的链表,与前面的e = p.next组合,可以遍历链表
                p = e;
            }
        }
        // 表示在桶中找到key值、hash值与插入元素相等的结点
        if (e != null) { 
            // 记录e的value
            V oldValue = e.value;
            // onlyIfAbsent为false或者旧值为null
            if (!onlyIfAbsent || oldValue == null)
                //用新值替换旧值
                e.value = value;
            // 访问后回调
            afterNodeAccess(e);
            // 返回旧值
            return oldValue;
        }
    }
    // 结构性修改
    ++modCount;
    // 实际大小大于阈值则扩容
    if (++size > threshold)
        resize();
    // 插入后回调
    afterNodeInsertion(evict);
    return null;
} 

get方法

public V get(Object key) {
    Node<K,V> e;
    return (e = getNode(hash(key), key)) == null ? null : e.value;
}

final Node<K,V> getNode(int hash, Object key) {
    Node<K,V>[] tab; Node<K,V> first, e; int n; K k;
    if ((tab = table) != null && (n = tab.length) > 0 &&
        (first = tab[(n - 1) & hash]) != null) {
        // 数组元素相等
        if (first.hash == hash && // always check first node
            ((k = first.key) == key || (key != null && key.equals(k))))
            return first;
        // 桶中不止一个节点
        if ((e = first.next) != null) {
            // 在树中get
            if (first instanceof TreeNode)
                return ((TreeNode<K,V>)first).getTreeNode(hash, key);
            // 在链表中get
            do {
                if (e.hash == hash &&
                    ((k = e.key) == key || (key != null && key.equals(k))))
                    return e;
            } while ((e = e.next) != null);
        }
    }
    return null;
}

resize方法

进行扩容,会伴随着一次重新hash分配,并且会遍历hash表中所有的元素,是非常耗时的。在编写程序中,要尽量避免resize。

final Node<K,V>[] resize() {
    Node<K,V>[] oldTab = table;
    int oldCap = (oldTab == null) ? 0 : oldTab.length;
    int oldThr = threshold;
    int newCap, newThr = 0;
    if (oldCap > 0) {
        // 超过最大值就不再扩充了,就只好随你碰撞去吧
        if (oldCap >= MAXIMUM_CAPACITY) {
            threshold = Integer.MAX_VALUE;
            return oldTab;
        }
        // 没超过最大值,就扩充为原来的2倍
        else if ((newCap = oldCap << 1) < MAXIMUM_CAPACITY && oldCap >= DEFAULT_INITIAL_CAPACITY)
            newThr = oldThr << 1; // double threshold
    }
    else if (oldThr > 0) // initial capacity was placed in threshold
        newCap = oldThr;
    else { 
        signifies using defaults
        newCap = DEFAULT_INITIAL_CAPACITY;
        newThr = (int)(DEFAULT_LOAD_FACTOR * DEFAULT_INITIAL_CAPACITY);
    }
    // 计算新的resize上限
    if (newThr == 0) {
        float ft = (float)newCap * loadFactor;
        newThr = (newCap < MAXIMUM_CAPACITY && ft < (float)MAXIMUM_CAPACITY ? (int)ft : Integer.MAX_VALUE);
    }
    threshold = newThr;
    @SuppressWarnings({"rawtypes","unchecked"})
        Node<K,V>[] newTab = (Node<K,V>[])new Node[newCap];
    table = newTab;
    if (oldTab != null) {
        // 把每个bucket都移动到新的buckets中
        for (int j = 0; j < oldCap; ++j) {
            Node<K,V> e;
            if ((e = oldTab[j]) != null) {
                oldTab[j] = null;
                if (e.next == null)
                    newTab[e.hash & (newCap - 1)] = e;
                else if (e instanceof TreeNode)
                    ((TreeNode<K,V>)e).split(this, newTab, j, oldCap);
                else { 
                    Node<K,V> loHead = null, loTail = null;
                    Node<K,V> hiHead = null, hiTail = null;
                    Node<K,V> next;
                    do {
                        next = e.next;
                        // 原索引
                        if ((e.hash & oldCap) == 0) {
                            if (loTail == null)
                                loHead = e;
                            else
                                loTail.next = e;
                            loTail = e;
                        }
                        // 原索引+oldCap
                        else {
                            if (hiTail == null)
                                hiHead = e;
                            else
                                hiTail.next = e;
                            hiTail = e;
                        }
                    } while ((e = next) != null);
                    // 原索引放到bucket里
                    if (loTail != null) {
                        loTail.next = null;
                        newTab[j] = loHead;
                    }
                    // 原索引+oldCap放到bucket里
                    if (hiTail != null) {
                        hiTail.next = null;
                        newTab[j + oldCap] = hiHead;
                    }
                }
            }
        }
    }
    return newTab;
}

HashMap常用方法测试

package map;

import java.util.Collection;
import java.util.HashMap;
import java.util.Set;

public class HashMapDemo {

    public static void main(String[] args) {
        HashMap<String, String> map = new HashMap<String, String>();
        // 键不能重复,值可以重复
        map.put("san", "张三");
        map.put("si", "李四");
        map.put("wu", "王五");
        map.put("wang", "老王");
        map.put("wang", "老王2");// 老王被覆盖
        map.put("lao", "老王");
        System.out.println("-------直接输出hashmap:-------");
        System.out.println(map);
        /**
         * 遍历HashMap
         */
        // 1.获取Map中的所有键
        System.out.println("-------foreach获取Map中所有的键:------");
        Set<String> keys = map.keySet();
        for (String key : keys) {
            System.out.print(key+"  ");
        }
        System.out.println();//换行
        // 2.获取Map中所有值
        System.out.println("-------foreach获取Map中所有的值:------");
        Collection<String> values = map.values();
        for (String value : values) {
            System.out.print(value+"  ");
        }
        System.out.println();//换行
        // 3.得到key的值的同时得到key所对应的值
        System.out.println("-------得到key的值的同时得到key所对应的值:-------");
        Set<String> keys2 = map.keySet();
        for (String key : keys2) {
            System.out.print(key + ":" + map.get(key)+"   ");

        }
        /**
         * 另外一种不常用的遍历方式
         */
        // 当我调用put(key,value)方法的时候,首先会把key和value封装到
        // Entry这个静态内部类对象中,把Entry对象再添加到数组中,所以我们想获取
        // map中的所有键值对,我们只要获取数组中的所有Entry对象,接下来
        // 调用Entry对象中的getKey()和getValue()方法就能获取键值对了
        Set<java.util.Map.Entry<String, String>> entrys = map.entrySet();
        for (java.util.Map.Entry<String, String> entry : entrys) {
            System.out.println(entry.getKey() + "--" + entry.getValue());
        }
        
        /**
         * HashMap其他常用方法
         */
        System.out.println("after map.size():"+map.size());
        System.out.println("after map.isEmpty():"+map.isEmpty());
        System.out.println(map.remove("san"));
        System.out.println("after map.remove():"+map);
        System.out.println("after map.get(si):"+map.get("si"));
        System.out.println("after map.containsKey(si):"+map.containsKey("si"));
        System.out.println("after containsValue(李四):"+map.containsValue("李四"));
        System.out.println(map.replace("si", "李四2"));
        System.out.println("after map.replace(si, 李四2):"+map);
    }

}

*12. 简述ConcurrentHashMap 的工作原理及分析源码(1.7版本)

思路:先介绍HashMap和HashTable在线程安全方面的问题,再介绍ConcurrentHashMap在线程安全方面的优势,采用了什么技术。

ConcurrentHashMap是Java并发包中提供的一个线程安全且高效的HashMap实现,ConcurrentHashMap在并发编程的场景中使用频率非常之高,本文就来分析下ConcurrentHashMap的实现原理,并对其实现原理进行分析(JDK1.7).

ConcurrentHashMap实现原理

  众所周知,哈希表是种非常高效,复杂度为O(1)的数据结构,在Java开发中,我们最常见到最频繁使用的就是HashMap和HashTable,但是在线程竞争激烈的并发场景中使用都不够合理。

  HashMap :先说HashMap,HashMap是线程不安全的,在并发环境下,可能会形成环状链表(扩容时可能造成,具体原因自行百度google或查看源码分析),导致get操作时,cpu空转,所以,在并发环境中使用HashMap是非常危险的。

  HashTable : HashTable和HashMap的实现原理几乎一样,差别无非是1.HashTable不允许key和value为null;2.HashTable是线程安全的。但是HashTable线程安全的策略实现代价却太大了,简单粗暴,get/put所有相关操作都是synchronized的,这相当于给整个哈希表加了一把大锁,多线程访问时候,只要有一个线程访问或操作该对象,那其他线程只能阻塞,相当于将所有的操作串行化,在竞争激烈的并发场景中性能就会非常差。

img

  HashTable性能差主要是由于所有操作需要竞争同一把锁,而如果容器中有多把锁,每一把锁锁一段数据,这样在多线程访问时不同段的数据时,就不会存在锁竞争了,这样便可以有效地提高并发效率。这就是ConcurrentHashMap所采用的”分段锁“思想。

  img

ConcurrentHashMap源码分析(1.7版本)

ConcurrentHashMap采用了非常精妙的”分段锁”策略,ConcurrentHashMap的主干是个Segment数组。

 final Segment<K,V>[] segments;

  Segment继承了ReentrantLock,所以它就是一种可重入锁(ReentrantLock)。在ConcurrentHashMap,一个Segment就是一个子哈希表,Segment里维护了一个HashEntry数组,并发环境下,对于不同Segment的数据进行操作是不用考虑锁竞争的。(就按默认的ConcurrentLeve为16来讲,理论上就允许16个线程并发执行,有木有很酷)

  所以,对于同一个Segment的操作才需考虑线程同步,不同的Segment则无需考虑。

Segment类似于HashMap,一个Segment维护着一个HashEntry数组

 transient volatile HashEntry<K,V>[] table;

HashEntry是目前我们提到的最小的逻辑处理单元了。一个ConcurrentHashMap维护一个Segment数组,一个Segment维护一个HashEntry数组。

 static final class HashEntry<K,V> {
        final int hash;
        final K key;
        volatile V value;
        volatile HashEntry<K,V> next;
        //其他省略
}    

我们说Segment类似哈希表,那么一些属性就跟我们之前提到的HashMap差不离,比如负载因子loadFactor,比如阈值threshold等等,看下Segment的构造方法

Segment(float lf, int threshold, HashEntry<K,V>[] tab) {
            this.loadFactor = lf;//负载因子
            this.threshold = threshold;//阈值
            this.table = tab;//主干数组即HashEntry数组
        }

我们来看下ConcurrentHashMap的构造方法

 1  public ConcurrentHashMap(int initialCapacity,float loadFactor, int concurrencyLevel) {
 3           if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0)
 4               throw new IllegalArgumentException();
 5           //MAX_SEGMENTS 为1<<16=65536,也就是最大并发数为65536
 6           if (concurrencyLevel > MAX_SEGMENTS)
 7               concurrencyLevel = MAX_SEGMENTS;
 8           //2的sshif次方等于ssize,例:ssize=16,sshift=4;ssize=32,sshif=5
 9          int sshift = 0;
10          //ssize 为segments数组长度,根据concurrentLevel计算得出
11          int ssize = 1;
12          while (ssize < concurrencyLevel) {
13              ++sshift;
14              ssize <<= 1;
15          }
16          //segmentShift和segmentMask这两个变量在定位segment时会用到,后面会详细讲
17          this.segmentShift = 32 - sshift;
18          this.segmentMask = ssize - 1;
19          if (initialCapacity > MAXIMUM_CAPACITY)
20              initialCapacity = MAXIMUM_CAPACITY;
21          //计算cap的大小,即Segment中HashEntry的数组长度,cap也一定为2的n次方.
22          int c = initialCapacity / ssize;
23          if (c * ssize < initialCapacity)
24              ++c;
25          int cap = MIN_SEGMENT_TABLE_CAPACITY;
26          while (cap < c)
27              cap <<= 1;
28          //创建segments数组并初始化第一个Segment,其余的Segment延迟初始化
29          Segment<K,V> s0 =
30              new Segment<K,V>(loadFactor, (int)(cap * loadFactor),
31                               (HashEntry<K,V>[])new HashEntry[cap]);
32          Segment<K,V>[] ss = (Segment<K,V>[])new Segment[ssize];
33          UNSAFE.putOrderedObject(ss, SBASE, s0); 
34          this.segments = ss;
35      }

  初始化方法有三个参数,如果用户不指定则会使用默认值,initialCapacity为16,loadFactor为0.75(负载因子,扩容时需要参考),concurrentLevel为16。

  从上面的代码可以看出来,Segment数组的大小ssize是由concurrentLevel来决定的,但是却不一定等于concurrentLevel,ssize一定是大于或等于concurrentLevel的最小的2的次幂。比如:默认情况下concurrentLevel是16,则ssize为16;若concurrentLevel为14,ssize为16;若concurrentLevel为17,则ssize为32。为什么Segment的数组大小一定是2的次幂?其实主要是便于通过按位与的散列算法来定位Segment的index。

  接下来,我们来看看put方法

 public V put(K key, V value) {
        Segment<K,V> s;
        //concurrentHashMap不允许key/value为空
        if (value == null)
            throw new NullPointerException();
        //hash函数对key的hashCode重新散列,避免差劲的不合理的hashcode,保证散列均匀
        int hash = hash(key);
        //返回的hash值无符号右移segmentShift位与段掩码进行位运算,定位segment
        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);
    }

 从源码看出,put的主要逻辑也就两步:1.定位segment并确保定位的Segment已初始化 2.调用Segment的put方法。

 关于segmentShift和segmentMask

  segmentShift和segmentMask这两个全局变量的主要作用是用来定位Segment,int j =(hash >>> segmentShift) & segmentMask。

  segmentMask:段掩码,假如segments数组长度为16,则段掩码为16-1=15;segments长度为32,段掩码为32-1=31。这样得到的所有bit位都为1,可以更好地保证散列的均匀性

  segmentShift:2的sshift次方等于ssize,segmentShift=32-sshift。若segments长度为16,segmentShift=32-4=28;若segments长度为32,segmentShift=32-5=27。而计算得出的hash值最大为32位,无符号右移segmentShift,则意味着只保留高几位(其余位是没用的),然后与段掩码segmentMask位运算来定位Segment。

  get/put方法

  get方法

 public V get(Object key) {
        Segment<K,V> s; 
        HashEntry<K,V>[] tab;
        int h = hash(key);
        long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;        //先定位Segment,再定位HashEntry
        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;
    }

  get方法无需加锁,由于其中涉及到的共享变量都使用volatile修饰,volatile可以保证内存可见性,所以不会读取到过期数据。

  来看下concurrentHashMap代理到Segment上的put方法,Segment中的put方法是要加锁的。只不过是锁粒度细了而已。

final V put(K key, int hash, V value, boolean onlyIfAbsent) {
            HashEntry<K,V> node = tryLock() ? null :
                scanAndLockForPut(key, hash, value);//tryLock不成功时会遍历定位到的HashEnry位置的链表(遍历主要是为了使CPU缓存链表),若找不到,则创建HashEntry。tryLock一定次数后(MAX_SCAN_RETRIES变量决定),则lock。若遍历过程中,由于其他线程的操作导致链表头结点变化,则需要重新遍历。
            V oldValue;
            try {
                HashEntry<K,V>[] tab = table;
                int index = (tab.length - 1) & hash;//定位HashEntry,可以看到,这个hash值在定位Segment时和在Segment中定位HashEntry都会用到,只不过定位Segment时只用到高几位。
                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))) {
                            oldValue = e.value;
                            if (!onlyIfAbsent) {
                                e.value = value;
                                ++modCount;
                            }
                            break;
                        }
                        e = e.next;
                    }
                    else {
                        if (node != null)
                            node.setNext(first);
                        else
                            node = new HashEntry<K,V>(hash, key, value, first);
                        int c = count + 1;              //若c超出阈值threshold,需要扩容并rehash。扩容后的容量是当前容量的2倍。这样可以最大程度避免之前散列好的entry重新散列,具体在另一篇文章中有详细分析,不赘述。扩容并rehash的这个过程是比较消耗资源的。
                        if (c > threshold && tab.length < MAXIMUM_CAPACITY)
                            rehash(node);
                        else
                            setEntryAt(tab, index, node);
                        ++modCount;
                        count = c;
                        oldValue = null;
                        break;
                    }
                }
            } finally {
                unlock();
            }
            return oldValue;
        }

总结

  ConcurrentHashMap作为一种线程安全且高效的哈希表的解决方案,尤其其中的”分段锁”的方案,相比HashTable的全表锁在性能上的提升非常之大。

*13. 介绍 JDK1.8 的ConcurrentHashMap

01 做了什么改进?

在 JDK1.7 版本中,ConcurrentHashMap 由数组 + Segment + 分段锁实现,其内部分为一个个段(Segment)数组,Segment 通过继承 ReentrantLock 来进行加锁,通过每次锁住一个 segment 来降低锁的粒度而且保证了每个 segment 内地操作的线程安全性,从而实现全局线程安全。

但是这么做的缺陷就是每次通过 hash 确认位置时需要 2 次才能定位到当前 key 应该落在哪个槽:

  1. 通过 hash 值和 段数组长度-1 进行位运算确认当前 key 属于哪个段,即确认其在 segments 数组的位置。
  2. 再次通过 hash 值和 table 数组(即 ConcurrentHashMap 底层存储数据的数组)长度 - 1进行位运算确认其所在桶。

为了进一步优化性能,在 jdk1.8 版本中,对 ConcurrentHashMap 做了优化,取消了分段锁的设计,取而代之的是通过 cas 操作和 synchronized 关键字来实现优化,而扩容的时候也利用了一种分而治之的思想来提升扩容效率,在 JDK1.8 中 ConcurrentHashMap 的存储结构和 HashMap 基本一致,如下图所示:

img

02 为什么 key 和 value 不允许为 null?

在 HashMap 中,key 和 value 都是可以为 null 的,但是在 ConcurrentHashMap 中却不允许,这是为什么呢?

作者 Doug Lea 本身对这个问题有过回答,在并发编程中,null 值容易引来歧义, 假如先调用 get(key) 返回的结果是 null,那么我们无法确认是因为当时这个 key 对应的 value 本身放的就是 null,还是说这个 key 值根本不存在,这会引起歧义,如果在非并发编程中,可以进一步通过调用 containsKey 方法来进行判断,但是并发编程中无法保证两个方法之间没有其他线程来修改 key 值,所以就直接禁止了 null 值的存在。

而且作者 Doug Lea 本身也认为,假如允许在集合,如 map 和 set 等存在 null 值的话,即使在非并发集合中也有一种公开允许程序中存在错误的意思,这也是 Doug Lea 和 Josh Bloch(HashMap作者之一) 在设计问题上少数不同意见之一,而 ConcurrentHashMap 是 Doug Lea 一个人开发的,所以就直接禁止了 null 值的存在。

03 ConcurrentHashMap 如何保证线程的安全性?

在 ConcurrentHashMap 中,采用了大量的分而治之的思想来降低锁的粒度,提升并发性能。其源码中大量使用了CAS操作来保证安全性,而不是和 HashTable 一样,不论什么方法,直接简单粗暴的使用 synchronized关键字来实现,接下来的原理分析中,部分和 HashMap 类似之处本文就不再重复,本文主要从安全性方面来分析 ConcurrentHashMap 的设计。

03 - 1如何用 CAS 保证数组初始化的安全?

下面就是初始化的方法:

这里面有一个非常重要的变量 sizeCtl,这个变量对理解整个 ConcurrentHashMap 的原理非常重要。

sizeCtl 有四个含义:

  • sizeCtl<-1 表示有 N-1 个线程正在执行扩容操作,如 -2 就表示有 2-1 个线程正在扩容。
  • sizeCtl=-1 占位符,表示当前正在初始化数组。
  • sizeCtl=0 默认状态,表示数组还没有被初始化。
  • sizeCtl>0 记录下一次需要扩容的大小。

知道了这个变量的含义,上面的方法就好理解了,第二个分支采用了 CAS 操作,因为 SIZECTL 默认为 0,所以这里如果可以替换成功,则当前线程可以执行初始化操作,CAS 失败,说明其他线程抢先一步把 sizeCtl 改为了 -1。扩容成功之后会把下一次扩容的阈值赋值给 sc,即 sizeClt。

03-1 put 操作如何保证数组元素的可见性?

ConcurrentHashMap 中存储数据采用的 Node 数组是采用了 volatile 来修饰的,但是这只能保证数组的引用在不同线程之间是可用的,并不能保证数组内部的元素在各个线程之间也是可见的,所以这里我们判定某一个下标是否有元素,并不能直接通过下标来访问,那么应该如何访问呢?

put时判断下标是否有元素

可以看到,这里是通过 tabAt 方法来获取元素,而 tableAt 方法实际上就是一个 CAS 操作:

如果发现当前节点元素为空,也是通过 CAS 操作(casTabAt)来存储当前元素。

如果当前节点元素不为空,则会使用 synchronized 关键字锁住当前节点(数组中的节点),并进行对应的设值 操作:

精妙的计数方式

在HashMap中,调用 put 方法之后会通过 ++size 的方式来存储当前集合中元素的个数,但是在并发模式下,这种操作是不安全的,所以不能通过这种方式,那么是否可以通过 CAS 操作来修改 size 呢?

直接通过 CAS 操作来修改 size 是可行的,但是假如同时有非常多的线程要修改 size 操作,那么只会有一个线程能够替换成功,其他线程只能不断地尝试 CAS,这会影响到 ConcurrentHashMap 集合的性能,所以作者就想到了一个分而治之的思想来完成计数。

作者定义了一个数组来计数,而且这个用来计数的数组也能扩容,每次线程需要计数的时候,都通过随机的方式获取一个数组下标的位置进行操作,这样就可以尽可能的降低了锁的粒度,最后获取 size 时,则通过遍历数组来实现计数:

//用来计数的数组,大小为2的N次幂,默认为2
private transient volatile CounterCell[] counterCells;
@sun.misc.Contended static final class CounterCell {//数组中的对象
        volatile long value;//存储元素个数
        CounterCell(long x) { value = x; }
    }

addCount 计数方法

接下来我们看看 addCount 方法:

addCount

首先会判断 CounterCell 数组是不是为空,需要这里的是,这里的 CAS 操作是将 BASECOUNT 和 baseCount 进行比较,如果相等,则说明当前没有其他线程过来修改 baseCount(即 CAS 操作成功),此时则不需要使用 CounterCell 数组,而直接采用 baseCount 来计数。

假如 CounterCell 为空且 CAS 失败,那么就会通过调用 fullAddCount 方法来对 CounterCell 数组进行初始化。

fullAddCount 方法

这个方法也很长,看起来比较复杂,里面包含了对 CounterCell 数组的初始化和赋值等操作。

(1)初始化 CounterCell 数组

我们先不管,直接进入初始化的逻辑:

img

这里面有一个比较重要的变量 cellsBusy,默认是 0,表示当前没有线程在初始化或者扩容,所以这里判断如果 cellsBusy==0,而 as 其实在前面就是把全局变量 CounterCell 数组的赋值,这里之所以再判断一次就是再确认有没有其他线程修改过全局数组 CounterCell,所以条件满足的话就会通过 CAS 操作修改 cellsBusy 为 1,表示当前自己在初始化了,其他线程就不能同时进行初始化操作了。

最后可以看到,默认是一个长度为 2 的数组,也就是采用了 2 个数组位置进行存储当前 ConcurrentHashMap 的元素数量。

(2)CounterCell 如何赋值

初始化完成之后,如果再次调用 put 方法,那么就会进入 fullAddCount 方法的另一个分支:

这里面首先判断了 CounterCell 数组不为空,然后会再次判断数组中的元素是不是为空,因为如果元素为空,就需要初始化一个 CounterCell 对象放到数组,而如果元素不为空,则只需要 CAS 操作替换元素中的数量即可。

所以这里面的逻辑也很清晰,初始化 CounterCell 对象的时候也需要将 cellBusy 由 0 改成 1。

(3)计数数组 CounterCell 也能扩容吗?

最后我们再继续看其他分支:

主要看上图红框中的分支,一旦会进入这个分支,就说明前面所有分支都不满足,即:

  • 当前 CounterCell 数组已经初始化完成。
  • 当前通过 hash 计算出来的 CounterCell 数组下标中的元素不为 null。
  • 直接通过 CAS 操作修改 CounterCell 数组中指定下标位置中对象的数量失败,说明有其他线程在竞争修改同一个数组下标中的元素。
  • 当前操作不满足不允许扩容的条件。
  • 当前没有其他线程创建了新的 CounterCell 数组,且当前 CounterCell 数组的大小仍然小于 CPU 数量。

所以接下来就需要对 CounterCell 数组也进行扩容,这个扩容的方式和 ConcurrentHashMap 的扩容一样,也是将原有容量乘以 2,所以其实 CounterCell 数组的容量也是满足 2 的 N 次幂

参考资料

https://juejin.cn/post/6961288808742518798


文章作者: 银色回廊
版权声明: 本博客所有文章除特別声明外,均采用 CC BY 4.0 许可协议。转载请注明来源 银色回廊 !
评论
  目录