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

大家好，又见面了，我是你们的朋友全栈君。

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

ConcurrentHashMap与HashTable区别？

HashTable
put()源代码
这里写图片描述

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

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

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

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

 final Segment<K,V>[] segments;

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

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

synchronized 是一个同步锁 synchronized （）
- 同步锁当一个线程A 访问【资源】的代码同步块时，A线程就会持续持有当前锁的状态，如果其他线程B-E 也要访问【资源】时，此时代码同步块将会阻塞，因此B-E线程需要排队等待A线程释放锁的状态后才可以持有【资源】。
ReentrantLock 可重入锁 JDK 5 引入的。顾名思义可重复进入的锁【Re-entrantLock】，它表示该锁能支持一个线程对资源的重复加锁。同时还支持获取锁的公平和非公平性选择。
我们来说说他的两类特性：公平性、非公平性
简单的来说就是ReentrantLock构造对象的时候传入一个值有选择的判断他是否是公平还是不公平。如图例子
- 公平性：根据线程请求锁的顺序依次获取锁，当一个线程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数组
        }

构造方法

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

loadFactor 负载因子 0.75
threshold 初始容量 16
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()

Put 时候，通过Hash函数将即将要put 的元素均匀的放到所需要的Segment 段中，然后调用Segment的put 方法进行添加数据。
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();  
        }  
    }

调用Segment 的remove 方法，先定位当前要删除的元素C，此时需要把A、B元素全部复制一遍，一个一个接入到D上。
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 不存在可见性。

发布者：全栈程序员-用户IM，转载请注明出处：https://javaforall.cn/150260.html原文链接：https://javaforall.cn

【正版授权，激活自己账号】： Jetbrains全家桶Ide使用，1年售后保障，每天仅需1毛

【官方授权正版激活】： 官方授权正版激活支持Jetbrains家族下所有IDE 使用个人JB账号...

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

ConcurrentHashMap与HashTable区别？

volatile 变量

相关推荐

jenkins自定义构建参数_git提交到远程分支

windows下cmd查看端口占用情况[通俗易懂]

世纪之战：Python和Java，到底学哪个更好？

SpringBoot集成Mybatis保姆级教程（完整版）

Google资深工程师深度讲解Go语言-测试与性能调优(八)

解决VMware虚拟机桥接模式无法上网「建议收藏」

发表回复