经验首页 前端设计 程序设计 Java相关 移动开发 数据库/运维 软件/图像 大数据/云计算 其他经验
当前位置:技术经验 » Java相关 » Java » 查看文章
HashMap很美好,但线程不安全怎么办?ConcurrentHashMap告诉你答案!
来源:cnblogs  作者:JavaBuild  时间:2024/3/4 10:21:03  对本文有异议

写在开头

《耗时2天,写完HashMap》这篇文章中,我们提到关于HashMap线程不安全的问题,主要存在如下3点风险:

风险1: put的时候导致元素丢失;如两个线程同时put,且key值相同的情况下,后一个线程put操作覆盖了前一个线程的操作,导致前一个线程的元素丢失。
风险2: put 和 get 并发时会导致 get 到 null;若一个线程的put操作触发了数组的扩容,这时另外一个线程去get,因为扩容的操作很耗时,这时有可能会卡死或者get到null。
风险3: 多线程下扩容会死循环;多线程下触发扩容时,因为前一个线程已经破坏了原有链表结构,后一个线程再去读取节点,进行链接的时候,很可能发生顺序错乱,从而形成一个环形链表,进而导致死循环。

Hashtable解决线程安全靠谱吗?

那我们怎么办呢?很多小伙伴可能第一时间想到了HashTable,因为它和HashMap拥有者相似的功能,底层也是基于哈希表实现,数组+链表构建,数组容量到达阈值后,同样会自动扩容,Hashtable 默认的初始大小为 11,之后每次扩充,容量变为原来的 2n+1。并且,Hashtable内部的方法几乎都是synchronized关键字修饰,保证了线程的安全

哇!这样一看,Hashtable简直是解决HashMap线程不安全的天选之子啊!但事实上,因为性能的问题,Hashtable已经在被废弃的边缘了,非常不建议在代码中使用它,原因如下接着往下看。
我们先写一个小小的测试类,来感受一下Hashtable的使用。

【代码示例1】

public class Test {
    public static void main(String[] args) {
        HashMap<Integer, String> map = new HashMap<>();
        map.put(1, "I");
        map.put(2, "love");
        map.put(3, "Java");

        Hashtable<Integer, String> hashtable = new Hashtable<>();
        hashtable.put(1, "JavaBuild");
        for (Map.Entry<Integer, String> entry : hashtable.entrySet()) {
            System.out.println(entry.getKey()+":"+entry.getValue());
        }
    }
}

输出:

1:JavaBuild

然后,我们跟入到put中的原来,去看看它的底层实现

【源码解析1】

 public synchronized V put(K key, V value) {
        // Make sure the value is not null
        if (value == null) {
            throw new NullPointerException();
        }

        // Makes sure the key is not already in the hashtable.
        Entry<?,?> tab[] = table;
        int hash = key.hashCode();
        int index = (hash & 0x7FFFFFFF) % tab.length;
        @SuppressWarnings("unchecked")
        Entry<K,V> entry = (Entry<K,V>)tab[index];
        for(; entry != null ; entry = entry.next) {
            if ((entry.hash == hash) && entry.key.equals(key)) {
                V old = entry.value;
                entry.value = value;
                return old;
            }
        }

        addEntry(hash, key, value, index);
        return null;
    }

通过这段源码我们能够发现
1、Hashtable哈希值的计算,并没有像HashMap那样重新计算,而是直接取key的hashCode()方法,这样一来它的扰动次数明显降低,hash的重合度更高;
2,index的位置计算中,Hashtable采用了%取余运算,而HashMap采用的是&运算,我们知道位运算直接对内存数据进行操作,不需要转成十进制,处理速度非常快,相比之下Hashtable的效率低下。
3,底层大部分的方法都是synchronized修饰,我们知道用synchronized 来保证线程安全的效率非常低下。当一个线程访问同步方法时,其他线程也访问同步方法,可能会进入阻塞或轮询状态,如使用 put 添加元素,另一个线程不能使用 put 添加元素,也不能使用 get,竞争会越来越激烈效率越低。

以上3点足以让我们头也不回的舍弃Hashtable,那么问题来了,除了这个集合类外,我们还有什么选项呢?这时,ConcurrentHashMap 高高的举起了它的小手!

ConcurrentHashMap

文章写到这些,终于引出了我们今天的主角,ConcurrentHashMap!作为一个效率又高,又能保证线程安全的集合类,它的使用频率非常之高,话不多说,我们先来画一个底层逻辑实现图感受一下它的魅力!
JDK1.8下的ConcurrentHashMap底层实现

哦,对了,虽然我们现在主流的Java版本都是1.8+了,但很多公司在面试的时候,提及ConcurrentHashMap时,有时候还是会问到1.7的底层实现,因此,学有余力的小伙伴,私下里把JDK1.7的底层源码也拿过来读读哈(build哥本地没有安装JDK1.7,就不贴源码解析了)。

JDK1.8中ConcurrentHashMap抛弃了原有的 Segment 分段锁,采用了 CAS + synchronized 来保证并发安全性,底层结构采用Node数组+链表/红黑树,当链表长度达到一定长度后,会转为红黑树,这和HashMap一样。

【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) {
    // key 和 value 不能为空
    if (key == null || value == null) throw new NullPointerException();
    int hash = spread(key.hashCode());
    int binCount = 0;
    for (Node<K,V>[] tab = table;;) {
        // f = 目标位置元素
        Node<K,V> f; int n, i, fh;// fh 后面存放目标位置的元素 hash 值
        if (tab == null || (n = tab.length) == 0)
            // 数组桶为空,初始化数组桶(自旋+CAS)
            tab = initTable();
        else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
            // 桶内为空,CAS 放入,不加锁,成功了就直接 break 跳出
            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;
            // 使用 synchronized 加锁加入节点
            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;
                            }
                        }
                    }
                    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;
  • 判断数组桶是否为空,若为空则通过tab = initTable(),初始化数组桶(自旋+CAS);
  • 计算出key的数组桶位置后,如果为空表示当前位置可以写入数据,利用 CAS 尝试写入,失败则自旋保证成功;
  • 如果当前位置的 “hashcode == MOVED == -1”,则需要进行扩容;
  • 如果都不满足,则利用 synchronized 锁写入数据;
  • 如果数量大于 TREEIFY_THRESHOLD 则要执行树化方法,在 treeifyBin 中会首先判断当前数组长度 ≥64 时才会将链表转换为红黑树。

【源码扩展1】
上面put的时候,若Node数组桶为空时,需要进行初始化,那么我们跟入initTable()中去看一看它的源码实现。

/**
 * Initializes table, using the size recorded in sizeCtl.
 */
private final Node<K,V>[] initTable() {
    Node<K,V>[] tab; int sc;
    while ((tab = table) == null || tab.length == 0) {
        // 如果 sizeCtl < 0 ,说明另外的线程执行CAS 成功,正在进行初始化。
        if ((sc = sizeCtl) < 0)
            // 让出 CPU 使用权
            Thread.yield(); // lost initialization race; just spin
        else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
            try {
                if ((tab = table) == null || tab.length == 0) {
                    int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
                    @SuppressWarnings("unchecked")
                    Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                    table = tab = nt;
                    sc = n - (n >>> 2);
                }
            } finally {
                sizeCtl = sc;
            }
            break;
        }
    }
    return tab;
}

从源码中我们可以看到,它的初始化是通过CAS和自旋完成的,注意其中的sizeCtl私有成员变量,当它的值小于0(准确来说等于-1)时,说明另外的线程执行CAS 成功,正在进行初始化。通过Thread.yield()做线程让步动作,让出CPU的使用权,自旋等待,随着获得资源,进入CAS。

知识点补充

CAS(compare and swap) 译为:比较与交换

// 如果在这个位置(address) 的值等于 这个值(expectedValue),那么交换(newValue)。
boolean CAS(address,expectedValue,newValue) {
	if(address 的 value == expectedValue) {
		address 的 value = newValue;
		return true;
	}
}

自旋: 所谓的自旋,旨在线程抢锁失败后进入阻塞状态,放弃 CPU,需要过很久才能再次被调度。但经过测算,大部分情况下,虽然当前抢锁失败,但过不了很久,锁就会被释放。因此,当某个线程抢占 CPU 失败后,保持就绪状态,一旦锁释放,就会继续抢占。
以上这2点内容,在后面的并发多线程中会着重学习,在这里浅浅点名,让大家明白他们的意思和作用即可。

【源码扩展2】
当链表的长度大于8时,会转为红黑树,而红黑树的实现,是通过底层的TreeBin,我们跟进去看一下。

static final class TreeBin<K,V> extends Node<K,V> {
        TreeNode<K,V> root;
        volatile TreeNode<K,V> first;
        volatile Thread waiter;
        volatile int lockState;
        // values for lockState
        static final int WRITER = 1; // set while holding write lock
        static final int WAITER = 2; // set when waiting for write lock
        static final int READER = 4; // increment value for setting read lock
...
}

TreeBin通过root属性维护红黑树的根结点,因为红黑树在旋转的时候,根结点可能会被它原来的子节点替换掉,在这个时间点,如果有其他线程要写这棵红黑树就会发生线程不安全问题,所以在 ConcurrentHashMap 中TreeBin通过waiter属性维护当前使用这棵红黑树的线程,来防止其他线程的进入。

【Get源码解析】
与put相比,get的源码就简单太多了,大概进行了如下几步操作:
1,根据计算出来的 hash 值寻址,如果在桶上直接返回值;
2,如果是红黑树,按照树的方式获取值;
3,如果是链表,按链表的方式遍历获取值;

public V get(Object key) {
    Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
    // key 所在的 hash 位置
    int h = spread(key.hashCode());
    if ((tab = table) != null && (n = tab.length) > 0 &&
        (e = tabAt(tab, (n - 1) & h)) != null) {
        // 如果指定位置元素存在,头结点hash值相同
        if ((eh = e.hash) == h) {
            if ((ek = e.key) == key || (ek != null && key.equals(ek)))
                // key hash 值相等,key值相同,直接返回元素 value
                return e.val;
        }
        else if (eh < 0)
            // 头结点hash值小于0,说明正在扩容或者是红黑树,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;
}

总结

文章写到这里,ConcurrentHashMap的介绍基本讲完了,我们现在来自我总结一下为啥它的效率又高,又能保证线程安全。
以JDK1.8版本阐述:

  1. Node + CAS + synchronized 保证并发安全,每次上锁的颗粒度细到链表或红黑树的根节点,不会影响其他Node的读写,此外CAS是轻量级的,synchronized 也经过了锁升级;
  2. JDK1.7的版本里采用的Segment 分段锁,颗粒度粗不说,Segment 的个数一旦初始化就不能改变。 Segment 数组的大小默认是 16,也就是说默认可以同时支持 16 个线程并发写。而1.8的版本中,Node是一个数组,初始默认为16,后续仍然可以以2的幂次方级别进行扩容,因此,它所支持的并发量要看它数组的真实容量;
  3. 效率高是因为它底层采用了和JDK1.8中HashMap相同的数组+链表/红黑树结构。

结尾彩蛋

如果本篇博客对您有一定的帮助,大家记得留言+点赞+收藏呀。原创不易,转载请联系Build哥!

如果您想与Build哥的关系更近一步,还可以关注俺滴公众号“JavaBuild888”,在这里除了看到《Java成长计划》系列博文,还有提升工作效率的小笔记、读书心得、大厂面经、人生感悟等等,欢迎您的加入!

原文链接:https://www.cnblogs.com/JavaBuild/p/18048664

 友情链接:直通硅谷  直通硅谷 怎么样 mac软件下载