Java岗大厂面试百日冲刺【Day19】—— 集合框架3

转载自: Java岗大厂面试百日冲刺【Day19】—— 集合框架3

本文已获得原作者 _陈哈哈 授权并经过重新整理规划后发布。

本栏目Java开发岗高频面试题主要出自以下各技术栈:Java基础知识、集合容器、并发编程、JVM、Spring全家桶、MyBatis等ORMapping框架、MySQL数据库、Redis缓存、RabbitMQ消息队列、Linux操作技巧等。

上回问到HashMap的线程安全问题,我们已经知道,在Java中有HashTable、SynchronizedMap、ConcurrentHashMap这三种是实现线程安全的Map。而ConcurrentHashMap也是最常用的并发场景下Map的选择,相信面试官对其理论和实战知识也是在熟悉不过,因此如果不能深入了解,或许会轻易被问住。

面试题1:先说一下大家为什么要选择ConcurrentHashMap?

在并发编程中使用HashMap可能导致程序死循环。而使用线程安全的HashTable效率又非常低下,基于以上两个原因,便有了ConcurrentHashMap的登场机会

1)线程不安全的HashMap

在多线程环境下,使用HashMap进行put操作会引起死循环,导致CPU利用率接近100%,所以在并发情况下不能使用HashMap。HashMap在并发执行put操作时会引起死循环,是因为多线程环境下会导致HashMap的Entry链表形成环形数据结构,一旦形成环形数据结构,Entry的next节点永远不为空,调用.next()时就会产生死循环获取Entry。

2)效率低下的HashTable

HashTable容器使用synchronized来保证线程安全,但在线程竞争激烈的情况下HashTable的效率非常低下(类似于数据库中的串行化隔离级别)。因为当一个线程访问HashTable的同步方法,其他线程也访问HashTable的同步方法时,会进入阻塞或轮询状态。如线程1使用put进行元素添加,线程2不但不能使用put方法添加元素,也不能使用get方法来获取元素,读写操作均需要获取锁,竞争越激烈效率越低。

因此,若未明确严格要求业务遵循串行化时(如转账、支付类业务),建议不启用HashTable。

InterviewForJavaByThreeQuestionsADay19-01.webp

3)ConcurrentHashMap的分段锁技术可有效提升并发访问率

HashTable容器在竞争激烈的并发环境下表现出效率低下的原因是所有访问HashTable的线程都必须竞争同一把锁,假如容器里有多把锁,每一把锁用于锁容器其中一部分数据,那么当多线程访问容器里不同数据段的数据时,线程间就不会存在严重锁竞争,从而可以有效提高并发访问效率,这就是ConcurrentHashMap所使用的分段锁技术。首先将数据分成一段一段地存储(一堆Segment),然后给每一段数据配一把锁,当一个线程占用锁访问其中一个段数据的时候,其他段的数据也能被其他线程访问。

对于 ConcurrentHashMap 你至少要知道的几个点:

  • 默认数组大小为16
  • 扩容因子为0.75,扩容后数组大小翻倍
  • 当存储的node总数量 >= 数组长度*扩容因子时,会进行扩容(数组中的元素、链表元素、红黑树元素都是内部类Node的实例或子类实例,这里的node总数量是指所有put进map的node数量)
  • 链表长度>=8数组长度<64时会进行扩容
  • 当数组下是链表时,在扩容的时候会从链表的尾部开始rehash
  • 链表长度>=8数组长度>=64时链表会变成红黑树
  • 树节点减少直至为空时会将对应的数组下标置空,下次存储操作再定位在这个下标t时会按照链表存储
  • 扩容时树节点数量<=6时会变成链表
  • 当一个事物操作发现map正在扩容时,会帮助扩容
  • map正在扩容时获取(get等类似操作)操作还没进行扩容的下标会从原来的table获取,扩容完毕的下标会从新的table中获取

面试题2:ConcurrentHashMap在JDK1.7、1.8中都有哪些优化?

其实,JDK1.8版本的ConcurrentHashMap的数据结构已经接近HashMap,相对而言,ConcurrentHashMap只是增加了同步的操作来控制并发。

  • JDK1.7:ReentrantLock+Segment+HashEntry
  • JDK1.8:Synchronized+CAS+Node(HashEntry)+红黑树

从JDK1.7版本的ReentrantLock+Segment+HashEntry,到JDK1.8版本中synchronized+CAS+HashEntry+红黑树。其中抛弃了原有的 Segment 分段锁,而采用了 CAS + synchronized 来保证并发安全性。

InterviewForJavaByThreeQuestionsADay19-02.webp

数据结构上跟HashMap很像,从1.7到1.8版本,由于HashEntry从链表 → 红黑树所以 concurrentHashMap的时间复杂度从O(n)到O(log(n)) ↓↓↓;

InterviewForJavaByThreeQuestionsADay19-03.webp

同时,也把之前的HashEntry改成了Node,作用不变,当Node链表的节点数大于8时Node会自动转化为TreeNode,会转换成红黑树的结构。把值和next采用了volatile去修饰,保证了可见性,并且也引入了红黑树,在链表大于一定值的时候会转换(默认是8)。

归纳一下:

  • JDK1.8的实现降低锁的粒度,JDK1.7版本锁的粒度是基于Segment的,包含多个HashEntry,而JDK1.8锁的粒度就是HashEntry(首节点)
  • JDK1.8版本的数据结构变得更加简单,使得操作也更加清晰流畅,因为已经使用synchronized来进行同步,所以不需要分段锁的概念(jdk1.8),也就不需要Segment这种数据结构了,由于粒度的降低,实现的复杂度也增加了
  • JDK1.8使用红黑树来优化链表,基于长度很长的链表的遍历是一个很漫长的过程,而红黑树的遍历效率是很快的,成功代替了一定阈值的链表。

追问1:JDK1.8为什么使用Synchronized来代替ReentrantLock?

JDK1.8为什么使用内置锁synchronized来代替重入锁ReentrantLock,主要有以下几点:

  1. 因为粒度降低了,在相对而言的低粒度加锁方式,synchronized并不比ReentrantLock差,在粗粒度加锁中ReentrantLock可能通过Condition来控制各个低粒度的边界,更加的灵活,而在低粒度中,Condition的优势就没有了
  2. JVM的开发团队从来都没有放弃synchronized,而且基于JVM的synchronized优化空间更大,使用内嵌的关键字比使用API更加自然
  3. 在大量的数据操作下,对于JVM的内存压力,基于API的ReentrantLock会开销更多的内存,虽然不是瓶颈,但是也是一个原因之一。

追问2:讲讲ConcurrentHashMap的 get put 过程?

JDK1.7版本的get put

在JDK1.7版本中,ConcurrentHashMap的数据结构是由一个Segment数组和多个HashEntry组成,如下图所示:

InterviewForJavaByThreeQuestionsADay19-04.webp

Segment数组的意义就是将一个大的table分割成多个小的table来进行加锁,也就是上面的提到的锁分段技术,而每一个Segment元素存储的是HashEntry数组+链表,这个和HashMap的数据存储结构一样。

初始化

ConcurrentHashMap的初始化是会通过位与运算来初始化Segment的大小,用ssize来表示,源码如下所示

private static final int DEFAULT_CONCURRENCY_LEVEL = 16;
private void writeObject(java.io.ObjectOutputStream s)
        throws java.io.IOException {
        // For serialization compatibility
        // Emulate segment calculation from previous version of this class
        int sshift = 0;
        int ssize = 1;
        while (ssize < DEFAULT_CONCURRENCY_LEVEL) {
            ++sshift;
            ssize <<= 1;
        }
        int segmentShift = 32 - sshift;
        int segmentMask = ssize - 1;

由此可以看出:因为ssize用位于运算来计算(ssize <<=1),所以Segment的大小取值都是以2的N次方,无关concurrencyLevel的取值,当然concurrencyLevel最大只能用16位的二进制来表示,即65536,换句话说,Segment的大小最多65536个,没有指定concurrencyLevel元素初始化,Segment的大小ssize默认为:DEFAULT_CONCURRENCY_LEVEL =16。

每一个Segment元素下的HashEntry的初始化也是按照位于运算来计算,用cap来表示,如下:

int cap = 1;
while (cap < c)
    cap <<= 1

如上所示,HashEntry大小的计算也是2的N次方(cap <<=1), cap的初始值为1,所以HashEntry最小的容量为2

JDK1.7 —— put操作

对于ConcurrentHashMap的数据插入,这里要进行两次Hash去定位数据的存储位置

static class Segment<K,V> extends ReentrantLock implements Serializable {
    private static final long serialVersionUID = 2249069246763182397L;
    final float loadFactor;
    Segment(float lf) { this.loadFactor = lf; }
}

从上Segment的继承体系可以看出,Segment实现了ReentrantLock,也就带有锁的功能,当执行put操作时,会进行第一次key的hash来定位Segment的位置,如果该Segment还没有初始化,即通过CAS操作进行赋值,然后进行第二次hash操作,找到相应的HashEntry的位置,这里会利用继承过来的锁的特性,在将数据插入指定的HashEntry位置时(链表的尾端),会通过继承 ReentrantLock 的 tryLock() 方法尝试去获取锁,如果获取成功就直接插入相应的位置,如果已经有线程获取该Segment的锁,那当前线程会以自旋的方式去继续的调用tryLock()方法去获取锁,超过指定次数就挂起,等待唤醒。

JDK1.7 —— get操作

ConcurrentHashMap的get操作跟HashMap类似,只是ConcurrentHashMap第一次需要经过一次hash定位到Segment的位置,然后再hash定位到指定的HashEntry,遍历该HashEntry下的链表进行对比,成功就返回,不成功就返回null

InterviewForJavaByThreeQuestionsADay19-05.webp

JDK1.8版本的get put

改进一:取消segments字段,直接采用transient volatile HashEntry<K,V>[] table保存数据,采用table数组元素作为锁,从而实现了对每一行数据进行加锁,进一步减少并发冲突的概率。

改进二:将原先table数组+单向链表的数据结构,变更为table数组+单向链表+红黑树的结构。

对于改进二的详细分析:

  1. 对于hash表来说,最核心的能力在于将key hash之后能均匀的分布在数组中。如果hash之后散列的很均匀,那么table数组中的每个队列长度基本都为0或者1才对。
  2. 但实际情况并非总是如此理想,虽然ConcurrentHashMap类默认的加载因子为0.75,但是在数据量过大或者运气不佳的情况下,还是会存在一些队列长度过长的情况,如果还是采用单向列表方式,那么查询某个节点的时间复杂度为O(n);
  3. 因此,对于个数超过8(默认值)的列表,jdk1.8中采用了红黑树的结构,那么查询的时间复杂度可以降低到O(logN),从而针对该种情况,改进了性能。

JDK1.8的实现已经摒弃了Segment的概念,而是直接用Node数组+链表+红黑树的数据结构来实现,并发控制使用Synchronized和CAS来操作,整个看起来就像是优化过且线程安全的HashMap,虽然在JDK1.8中还能看到Segment的数据结构,但是已经简化了属性,只是为了兼容旧版本。

在深入JDK1.8的put和get实现之前要知道一些常量设计和数据结构,这些是构成ConcurrentHashMap实现结构的基础,下面看一下基本属性:

// node数组最大容量:2^30=1073741824
private static final int MAXIMUM_CAPACITY = 1 << 30;
// 默认初始值,必须是2的幕数
private static final int DEFAULT_CAPACITY = 16
//数组可能最大值,需要与toArray()相关方法关联
static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
//并发级别,遗留下来的,为兼容以前的版本
private static final int DEFAULT_CONCURRENCY_LEVEL = 16;
// 负载因子
private static final float LOAD_FACTOR = 0.75f;
// 链表转红黑树阀值,> 8 链表转换为红黑树
static final int TREEIFY_THRESHOLD = 8;
//树转链表阀值,小于等于6(tranfer时,lc、hc=0两个计数器分别++记录原bin、新binTreeNode数量,<=UNTREEIFY_THRESHOLD 则untreeify(lo))
static final int UNTREEIFY_THRESHOLD = 6;
static final int MIN_TREEIFY_CAPACITY = 64;
private static final int MIN_TRANSFER_STRIDE = 16;
private static int RESIZE_STAMP_BITS = 16;
// 2^15-1,help resize的最大线程数
private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1;
// 32-16=16,sizeCtl中记录size大小的偏移量
private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;
// forwarding nodes的hash值
static final int MOVED     = -1;
// 树根节点的hash值
static final int TREEBIN   = -2;
// ReservationNode的hash值
static final int RESERVED  = -3;
// 可用处理器数量
static final int NCPU = Runtime.getRuntime().availableProcessors();
//存放node的数组
transient volatile Node<K,V>[] table;
/*控制标识符,用来控制table的初始化和扩容的操作,不同的值有不同的含义
 *当为负数时:-1代表正在初始化,-N代表有N-1个线程正在 进行扩容
 *当为0时:代表当时的table还没有被初始化
 *当为正数时:表示初始化或者下一次进行扩容的大小*/

基本属性定义了ConcurrentHashMap的一些边界以及操作时的一些控制,下面看一些内部的一些结构组成,这些是整个ConcurrentHashMap整个数据结构的核心。

InterviewForJavaByThreeQuestionsADay19-06.webp

Node

HashEntry == Node

Node是ConcurrentHashMap存储结构的基本单元,继承于HashMap中的Entry,用于存储数据,Node就是一个链表,但是只允许对数据进行查找,不允许进行修改;

TreeNode

TreeNode继承与Node,但是数据结构换成了二叉树结构,它是红黑树的数据的存储结构,用于红黑树中存储数据,当链表的节点数大于8时会转换成红黑树的结构,他就是通过TreeNode作为存储结构代替Node来转换成黑红树。源代码如下

TreeBin

TreeBin从字面含义中可以理解为存储树形结构的容器,而树形结构就是指TreeNode,所以TreeBin就是封装TreeNode的容器,它提供转换黑红树的一些条件和锁的控制。

现在通过一个简单的例子以debug的视角看看ConcurrentHashMap的具体操作细节

public class TestConcurrentHashMap{   
    public static void main(String[] args){
        ConcurrentHashMap<String,String> map = new ConcurrentHashMap(); //初始化ConcurrentHashMap
        //新增个人信息
        map.put("id","1");
        map.put("name","andy");
        map.put("sex","男");
        //获取姓名
        String name = map.get("name");
        Assert.assertEquals(name,"andy");
        //计算大小
        int size = map.size();
        Assert.assertEquals(size,3);
    }
}

我们先通过new ConcurrentHashMap()来进行初始化

public ConcurrentHashMap() {
}

由上你会发现ConcurrentHashMap的初始化其实是一个空实现,并没有做任何事,这里后面会讲到,这也是和其他的集合类有区别的地方,初始化操作并不是在构造函数实现的,而是在put操作中实现,当然ConcurrentHashMap还提供了其他的构造函数,有指定容量大小或者指定负载因子,跟HashMap一样。

JDK1.8 —— put操作

在上面的例子中我们新增个人信息会调用put方法,我们来看下

public V put(K key, V value) {
    return putVal(key, value, false);
}
/** Implementation for put and putIfAbsent */
final V putVal(K key, V value, boolean onlyIfAbsent) {
    if (key == null || value == null) throw new NullPointerException();
    int hash = spread(key.hashCode()); //两次hash,减少hash冲突,可以均匀分布
    int binCount = 0;
    for (Node<K,V>[] tab = table;;) { //对这个table进行迭代
        Node<K,V> f; int n, i, fh;
        //这里就是上面构造方法没有进行初始化,在这里进行判断,为null就调用initTable进行初始化,属于懒汉模式初始化
        if (tab == null || (n = tab.length) == 0)
            tab = initTable();
        else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {//如果i位置没有数据,就直接无锁插入
            if (casTabAt(tab, i, null,
                         new Node<K,V>(hash, key, value, null)))
                break;                   // no lock when adding to empty bin
        }
        else if ((fh = f.hash) == MOVED)//如果在进行扩容,则先进行扩容操作
            tab = helpTransfer(tab, f);
        else {
            V oldVal = null;
            //如果以上条件都不满足,那就要进行加锁操作,也就是存在hash冲突,锁住链表或者红黑树的头结点
            synchronized (f) {
                if (tabAt(tab, i) == f) {
                    if (fh >= 0) { //表示该节点是链表结构
                        binCount = 1;
                        for (Node<K,V> e = f;; ++binCount) {
                            K ek;
                            //这里涉及到相同的key进行put就会覆盖原先的value
                            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) { //如果链表的长度大于8时就会进行红黑树的转换
                if (binCount >= TREEIFY_THRESHOLD)
                    treeifyBin(tab, i);
                if (oldVal != null)
                    return oldVal;
                break;
            }
        }
    }
    addCount(1L, binCount);//统计size,并且检查是否需要扩容
    return null;
}

这个put的过程很清晰,对当前的table进行无条件自循环直到put成功,可以分成以下六步流程来概述:

  1. 如果没有初始化就先调用initTable()方法来进行初始化过程
  2. 如果没有hash冲突就直接CAS插入
  3. 如果还在进行扩容操作就先进行扩容
  4. 如果存在hash冲突,就加锁来保证线程安全,这里有两种情况,一种是链表形式就直接遍历到尾端插入,一种是红黑树就按照红黑树结构插入。
  5. 最后一个如果该链表的数量大于阈值8,就要先转换成黑红树的结构,break再一次进入循环,默认的链表大小,超过了这个值就会转换为红黑树;
  6. 如果添加成功就调用addCount()方法统计size,并且检查是否需要扩容。

put的流程你可以从中发现,他在并发处理中使用的是乐观锁,当有冲突的时候才进行并发处理。

JDK1.8 —— get操作

我们现在要回到开始的例子中,我们对个人信息进行了新增之后,我们要获取所新增的信息,使用 String name = map.get(“name”) 获取新增的 name 信息,现在我们依旧用debug的方式来分析下 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()); //计算两次hash
    if ((tab = table) != null && (n = tab.length) > 0 &&
        (e = tabAt(tab, (n - 1) & h)) != null) {//读取首节点的Node元素
        if ((eh = e.hash) == h) { //如果该节点就是首节点就返回
            if ((ek = e.key) == key || (ek != null && key.equals(ek)))
                return e.val;
        }
        //hash值为负值表示正在扩容,这个时候查的是ForwardingNode的find方法来定位到nextTable来
        //查找,查找到就返回
        else if (eh < 0)
            return (p = e.find(h, key)) != null ? p.val : null;
        while ((e = e.next) != null) {//既不是首节点也不是ForwardingNode,那就往下遍历
            if (e.hash == h &&
                ((ek = e.key) == key || (ek != null && key.equals(ek))))
                return e.val;
        }
    }
    return null;
}

ConcurrentHashMap 的 get 操作的流程很简单,也很清晰,可以分为三个步骤来描述

  1. 计算hash值,定位到该table索引位置,如果是首节点符合就返回
  2. 如果遇到扩容的时候,会调用标志正在扩容节点ForwardingNode的find方法,查找该节点,匹配就返回
  3. 以上都不符合的话,就往下遍历节点,匹配就返回,否则最后就返回null

追问3:ConcurrentHashMap 的 get 方法是否要加锁,为什么?

get 方法不需要加锁。因为 Node 的元素 value 和指针 next 是用 volatile 修饰的(可见性),在多线程环境下线程A修改节点的 value 或者新增节点的时候是对线程B可见的。

这也是它比其他并发集合比如 Hashtable、用 Collections.synchronizedMap()包装的 HashMap 效率高的原因之一。

面试题3:我们可以使用CocurrentHashMap来代替Hashtable吗?

我们知道Hashtable是synchronized的,但是ConcurrentHashMap同步性能更好,因为它仅仅根据同步级别对map的一部分进行上锁。ConcurrentHashMap当然可以代替HashTable,但是HashTable提供更强的线程安全性。它们都可以用于多线程的环境,但是当Hashtable的大小增加到一定的时候,性能会急剧下降,因为迭代时需要被锁定很长的时间。

因为ConcurrentHashMap引入了分割(segmentation),不论它变得多么大,仅仅需要锁定map的某个部分,而其它的线程不需要等到迭代完成才能访问map。简而言之,在迭代的过程中,ConcurrentHashMap仅仅锁定map的某个部分,而Hashtable则会锁定整个map

追问1:那ConcurrentHashMap有哪些缺陷?

ConcurrentHashMap 是设计为非阻塞的。在更新时会局部锁住某部分数据,但不会把整个表都锁住。同步读取操作则是完全非阻塞的。

  • 好处是:在保证合理的同步前提下,效率很高。
  • 坏处是:严格来说读取操作不能保证反映最近的更新。例如线程A调用putAll写入大量数据,期间线程B调用get,则只能get到目前为止已经顺利插入的部分数据,而未必是最新的数据。

因此,若需要严格按照串行事务定需求的话,如转账、支付类业务还是使用HashTable。

集合框架下一篇 (四)会继续沿着CocurrentHashMap深入讲解,包括CAS乐观锁原理、volatile、自旋锁等相关问题;

ConcurrentHashMap 不支持 key 或者 value 为 null 的原因?
Volatile 关键字干了那些事?Volatile的特性是什么?
不安全会导致哪些问题?如何解决?
有没有线程安全的并发容器?
ConcurrentHashMap并发度为啥好这么多?
CAS是啥?ABA是啥?场景有哪些,怎么解决?
自旋锁是什么?解决什么问题?
CAS性能很高,但是为什么jdk1.8之后还是会用Synchronized?
快速失败(fail-fast)是啥,应用场景有哪些?