欢迎您的访问
专注于分享最有价值的互联网技术干货

深入理解ConcurrentHashMap原理分析以及线程安全性问题

几个T的资料等你来白嫖
双倍快乐

在之前的文章提到ConcurrentHashMap是一个线程安全的,那么我么看一下ConcurrentHashMap如何进行操作的。

ConcurrentHashMap与HashTable区别?

HashTable
put()源代码

20210413134111554.png

20210413134112244.png
我们来看一下put 操作:
方法体 被 同步锁标记,由于同步锁的特性,其他线程将会排队进行等待处理。
除此之外,对传入的key 值进行了一个判断空值逻辑。【PS:HashMap 是允许key值为空的】

20210413134113664.png

ConcurrentHashMap

分段锁技术:ConcurrentHashMap 相比 HashTable 对锁的处理不同的点在于:前者是分段部分数据锁定
每一把锁用于锁容器其中一部分数据,那么当多线程访问容器里不同数据段的数据时,线程间就不会存在锁竞争,后者是全部锁定。【PS:下图 基于JDK 1.7绘制】
20210413134114073.png

20210413134114579.png

从图中可以看出来ConcurrentHashMap的主干是个Segment数组。

20210413134114851.png

由于ConConcurrentHashMap 的主体是由多个Segment 链式组成,因此每个Segment都持有自己的锁。

final Segment<K,V>[] segments;

需要注意的是 Segment 是一种可重入锁(继承ReentrantLock)

那么我简单说一下ReentrantLock 与synchronized有什么区别?

20210413134115385.png

  • synchronized 是一个同步锁 synchronized ()

    • 同步锁 当一个线程A 访问 【资源】的代码同步块时,A线程就会持续持有当前锁的状态,如果其他线程B-E 也要访问【资源】时,此时代码同步块将会阻塞,因此B-E线程需要排队等待A线程释放锁的状态后才可以持有【资源】。
  • ReentrantLock 可重入锁 JDK 5 引入的。 顾名思义 可重复进入的锁【Re-entrantLock】,它表示该锁能支持一个线程对资源的重复加锁。同时还支持获取锁的公平和非公平性选择。
  • 我们来说说他的两类特性:公平性、非公平性
    简单的来说就是ReentrantLock构造对象的时候传入一个值有选择的判断他是否是公平还是不公平。如图例子
  • 20210413134115666.png

    • 公平性:根据线程请求锁的顺序依次获取锁,当一个线程A 访问 【资源】的期间,线程A 获取锁资源,此时内部存在一个计数器num+1,在访问期间,线程B、C请求 资源时,发现A 线程在持有当前资源,因此在后面线程节点排队(B 处于待唤醒状态),假如此A线程再次请求资源时,不需要再次排队,可以直接再次获取当前资源 (内部计数器+1 num=2) ,当A线程释放所有锁的时候(num=0),此时会唤醒B线程进行获取锁的操作,其他C-E线程就同理。(情况2)
    • 非公平性:当A线程已经释放所之后,准备唤醒线程B获取资源时,此时线程M 获取请求,此时会出现竞争,线程B 没有竞争过M线程,测试M获取此线程,因此M会有限获得资源,B继续睡眠。(情况2)
  • synchronized 是一个非公平性锁。通俗来讲暴利等待。

非公平锁总体会比公平要好一些,它是根据每个线程对资源抢占能力来分配的,不需要严格的安装锁的请求顺序接入

ReentrantLock 使用场景

  • 定时任务 防止任务重复执行。
  • 套字节连接池,如果正在数据通信防止重复接入连接

在了解以上的功能 我们之后我们继续看一下ConcurrentHashMap核心构造方法代码。

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

构造方法

20210413134115947.png

从以上代码可以看出ConcurrentHashMap有比较重要的三个参数:

  1. loadFactor 负载因子 0.75
  2. threshold 初始 容量 16
  3. concurrencyLevel 实际上是Segment的实际数量 默认16。

ConcurrentHashMap如何发生ReHash?
ConcurrentLevel 一旦设定的话,就不会改变。ConcurrentHashMap当元素个数大于临界值的时候,就会发生扩容。但是ConcurrentHashMap与其他的HashMap不同的是,它不会对Segmengt 数量增大,只会增加Segmengt 后面的链表容量的大小。即对每个Segmengt 的元素进行的ReHash操作。


我们再看一下核心的ConcurrentHashMap Put ()方法:
1.7(版本)

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

final V put(K key, int hash, V value, boolean onlyIfAbsent) {
    HashEntry<K,V> node = tryLock() ? null :scanAndLockForPut(key, hash, value);
    //tryLock()是ReentrantLock获取锁一个方法。如果当前线程获取锁成功 返回true,如果别线程获取了锁返回false不成功时会遍历定位到的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;
        }

1.8版本

final V putVal(K key, V value, boolean onlyIfAbsent) {
        if (key == null || value == null) throw new NullPointerException();
        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) {
                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 (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;

1.7 的流程大致:
优先计算出Key 通过Hash函数计算出hash值 现计算出当前key属于哪个Segment 然后调用Segment.put 分段方法Segment.put()

  1. Put 时候 ,通过Hash函数将即将要put 的元素均匀的放到所需要的Segment 段中,然后调用Segment的put 方法进行添加数据。
  2. Segment的put 是加锁中完成的。如果当前元素数大于最大临界值的的话将会产生rehash. 先通过 getFirst 找到链表的表头部分,然后遍历链表,调用equals 比配是否存在相同的key ,如果找到的话,则将最新的Key 对应value值。如果没有找到,新增一个HashEntry 它加到整个Segment的头部。

1.8 采用红黑树的Node存储结构,因此在计算过程中做了一些调整优化。大致列了一下:

  • 由于是树形存储,计算hash值 进行spread 将散列的较高位散布(XOR)降低 将位数控制在int最大整数之内
  • 在添加数据元素过程中,将链表转换成树形结构,同时对树形结构节点元素查找和添加进行结构化的调整。
  • 写入数据中增加了CAS 方法 compareAndSwapInt,compareAndSwapLong

我们先看一下Get 方法的源码:
1.7

//计算Segment中元素的数量
transient volatile int count;
***********************************************************
    public V get(Object key) {  
        int hash = hash(key.hashCode());  
        return segmentFor(hash).get(key, hash);  
    }  
***********************************************************

    final Segment<K,V> segmentFor(int hash) {  
        return segments[(hash >>> segmentShift) & segmentMask];  
    }  
********************************************************
    V get(Object key, int hash) {  
        if (count != 0) { // read-volatile  
            HashEntry<K,V> e = getFirst(hash);  
            while (e != null) {  
                if (e.hash == hash && key.equals(e.key)) {  
                    V v = e.value;  
                    if (v != null)  
                        return v;  
                    return readValueUnderLock(e); // recheck  
                }  
                e = e.next;  
            }  
        }  
        return null;  
    }

1.7 大致流程如下:
1.传入读取Key值,通过Hash函数计算出 对应Segment 的位置。
2.调用segmentFor(int hash)方法,计算确定操作应该在哪一个segment中进行 ,通过右无符号位运算 进行右移操作 计算出偏移值 获得需要操作的Segment位置。

  • 确定需要操作的Segment后,再调用 get 方法获取对应的值。通过count 值先判断当前值是否为空。在调用getFirst()方法获取头节点,然后在Segment内部遍历列表通过equals对比的方式进行比对返回值。

1.8 Get 方法

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;

1.8 大致思路和1.7 一致,由于1.8存储结构形态采用了树形数据结构,还是通过get操作通过首先计算key的hash值来确定该元素放在数组的哪个位置,然后进行顺序遍历直到找到确定的数值。

ConcurrentHashMap为什么读的时候不加锁?

  • ConcurrentHashMap是分段并发分段进行读取数据的。
  • 【jdk1.7】 Segment 里面采用很多计数变量都通过volatile关键字修饰,由于volatile变量 happer-before的特性。导致get 方法能够几乎准确的获取最新的数据并且更新。

再看一下 1.7 jdk ConcurrentHashMapRemove()方法:

V remove(Object key, int hash, Object value) {  
        lock();  
        try {  
            int c = count - 1;  
            HashEntry<K,V>[] tab = table;  
            int index = hash & (tab.length - 1);  
            HashEntry<K,V> first = tab[index];  
            HashEntry<K,V> e = first;  
            while (e != null && (e.hash != hash || !key.equals(e.key)))  
                e = e.next;  

            V oldValue = null;  
            if (e != null) {  
                V v = e.value;  
                if (value == null || value.equals(v)) {  
                    oldValue = v;  
                    // All entries following removed node can stay  
                    // in list, but all preceding ones need to be  
                    // cloned.  
                    ++modCount;  
                    HashEntry<K,V> newFirst = e.next;  
                    for (HashEntry<K,V> p = first; p != e; p = p.next)  
                        newFirst = new HashEntry<K,V>(p.key, p.hash,  
                                                      newFirst, p.value);  
                    tab[index] = newFirst;  
                    count = c; // write-volatile  
                }  
            }  
            return oldValue;  
        } finally {  
            unlock();  
        }  
    }

20210413134116289.png

  1. 调用Segment 的remove 方法,先定位当前要删除的元素C,此时需要把A、B元素全部复制一遍,一个一个接入到D上。
  2. remove 也是在加锁的情况下进行的。
    PS:JDK1.8的删除元素在此就不详细展示,感兴趣的小伙伴可以私底下研究一下。

volatile 变量

volatile 变量 是保证修饰变量具有可见性的变量
我们发现 对于CurrentHashMap而言的话,源码里面又很多地方都用到了这个变量。比如HashEntry 、value 、Segment元素个数Count。

volatile 属于JMM 模型中的一个词语。首先先简单说一下 Java内存模型中的 几个概念:

  • 原子性:保证 Java内存模型中原子变量内存操作的。通常有 read、write、load、use、assign、store、lock、unlock等这些。
  • 可见性:就是当一个线程对一个变量进行了修改,其他线程即可立即得到这个变量最新的修改数据。
  • 有序性:如果在本线程内观察,所有操作都是有序的;如果在一个线程中观察另一个线程,所有操作都是无序的。
  • 先行发生:happen-before 先行发生原则是指Java内存模型中定义的两项操作之间的依序关系,如果说操作A先行发生于操作B,其实就是说发生操作B之前.
  • 传递性

volatile 变量 与普通变量的不同之处?

  • volatile 是有可见性,一定程度的有序性。
  • volatile 赋值的时候新值能够立即刷新到主内存中去,每次使用的时候能够立刻从内存中刷新。
    做一个简单例子看一下 这个功能
public class VolatileTest{

    int a=1;
    int b=2;

    //赋值操作
    public  void change(){
        a=3;
        b=a;
    }

    //打印操作
    public  void print(){
        System.out.println("b:"+b+",a:"+a);
    }

    @Test
    public void testNorMal(){
        VolatileTest vt=new VolatileTest();

     for (int i = 0; i < 100000; i++) {
         new Thread(new Runnable() {
                @Override
                public void run() {
                     try {
                        Thread.sleep(100);
                    } catch (InterruptedException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                     vt.change();
                }
            }).start();

            new Thread(new Runnable() {
                @Override
                public void run() {
                     try {
                        Thread.sleep(10);
                    } catch (InterruptedException e) {
                        // TODO Auto-generated catch block
                        e.printStackTrace();
                    }
                     vt.print();
                }
            }).start();
    }   

    }
}

跑了 n 次会出现一条 b=3,a=1 的错误打印记录。这就是因为普通变量相比volatile 不存在可见性。

赞(0) 打赏
版权归原创作者所有,任何形式转载请联系我们:大白菜博客 » 深入理解ConcurrentHashMap原理分析以及线程安全性问题

评论 抢沙发

3 + 5 =
  • 昵称 (必填)
  • 邮箱 (必填)
  • 网址

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

微信扫一扫打赏