浅谈ConcurrentHashMap实现原理

2018-09-10 01:02:46来源:博客园 阅读 ()

新老客户大回馈,云服务器低至5折

我们都知道HashTable是线程安全的类,因为使用了Synchronized来锁整张Hash表来实现线程安全,让线程独占;

ConcurrentHashMap的锁分离技术就是用多个锁来控制对Hash表的不同部分进行修改,因为我可能只需要对一小块部分进行操作,而如果锁整张表开销太大了,其内部实现就是用Semgent来控制的,每个Semgent都是一个小的HashTable,他们有自己的锁;

然后有些方法需要访问整个表,比如Size,ContainsValue肯定是要访问整个表的数据,这个时候就需要锁整张表,ConcurrentHashMap就会按顺序锁定每个Segment,操作完毕之后再按顺序释放每个Segment,按顺序是为了防止出现死锁;

引用一张图片解释ConcurrentHashMap的结构:

 

可以看出就是在HashMap的基础上多了一个数组(暂且这样理解),数组的每个元素就是Segment;Segment<K,V> extends ReentrantLock implements Serializable;

ConcurrentHashMap的构造函数:

 

public ConcurrentHashMap(int initialCapacity,
                             float loadFactor, int concurrencyLevel) {
        if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0)
            throw new IllegalArgumentException();
        if (concurrencyLevel > MAX_SEGMENTS)
            concurrencyLevel = MAX_SEGMENTS;
        // Find power-of-two sizes best matching arguments
        int sshift = 0;
        int ssize = 1;
        while (ssize < concurrencyLevel) {
            ++sshift;
            ssize <<= 1;
        }
        this.segmentShift = 32 - sshift;
        this.segmentMask = ssize - 1;
        if (initialCapacity > MAXIMUM_CAPACITY)
            initialCapacity = MAXIMUM_CAPACITY;
        int c = initialCapacity / ssize;
        if (c * ssize < initialCapacity)
            ++c;
        int cap = MIN_SEGMENT_TABLE_CAPACITY;
        while (cap < c)
            cap <<= 1;
        // create segments and segments[0]
        Segment<K,V> s0 =
            new Segment<K,V>(loadFactor, (int)(cap * loadFactor),
                             (HashEntry<K,V>[])new HashEntry[cap]);
        Segment<K,V>[] ss = (Segment<K,V>[])new Segment[ssize];
        UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]
        this.segments = ss;
    }

 

 

 

ConcurrentHahsMap进行Put操作的时候,对Key需要进行3次Hash运算才能确定最终的位置:hash1(key) -> x1(得到一个值);hash2(x1) -> x2(确定Segment位置,第二次hash是为了减少hash碰撞);hash3(x2) -> x3(确定元素放入哪个HashEntry);这里只是用比较好理解的白话说,下面会说原理;

 

ConcurrentHashMap中主要实体类就是三个:ConcurrentHashMap(整个Hash表),Segment(桶),HashEntry(节点);

 static final class HashEntry<K,V> {  
     final K key;  
     final int hash;  
     volatile V value;  
     final HashEntry<K,V> next;  
 } 

 

可以看出,除了value不是final,其next属性都是final,说明不能从hash链的中间或尾部添加或删除节点;对于Put操作,可以一律将元素添加到hash链的头部,而对于remove操作,从中间删除一个节点,将该节点的前面所有节点复制一份并指向删除节点的下一节点;这会产生两条Hash链,这也是为了保证多线程访问的同步性;将value设置成volatile,这样避免了加锁;因为一个线程在执行get,另一个线程在执行remove,remove还没有执行完的时候,get能看到remove之前的值,如下图:

 

下面说说Segment的数据结构:

static final class Segment<K,V> extends ReentrantLock implements Serializable {    
         /** 
          * count用来统计该段数据的个数,它是volatile,如增加/删除节点,都要写count值,每次读取操作开始都要读取count的值。
          */
         transient volatileint count;  
         /** 
          * modCount统计段结构改变的次数,主要是为了检测对多个段进行遍历过程中某个段是否发生改变
          */
         transient int modCount;  
         /** 
          * rehash界限值
          */
         transient int threshold;  
         /** 
          * 就是上面的HashEntry
          */
         transient volatile HashEntry<K,V>[] table;  
         /** 
          * 负载因子
          */
         final float loadFactor;  
 }

 

ConcurrentHashMap的get方法是直接调用Segment的get方法:

 

public V get(Object key) {
    int hash = hash(key); // throws NullPointerException if key null
    return segmentFor(hash).get(key, hash);
}

 

 

V get(Object key, int hash) {  
     if (count != 0) { // read-volatile 当前桶的数据个数是否为0 
         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;  
 } 

 

 V readValueUnderLock(HashEntry<K,V> e) {  
     lock();  
     try {  
         return e.value;  
     } finally {  
         unlock();  
     }  
 }

get方法是不需要加锁的,因为上面已经看到了count和HashEntry都是volatile;

源码中增加value可能为null的情况,是为了保险起见,当节点正在发生改变而又未锁定时就可能为空,而HashEntry中的value不是final的,所以个人认为是为了加此判断;

如用于统计当前Segement大小的count字段和用于存储值的HashEntry的value。定义成volatile的变量,能够在线程之间保持可见性,能够被多线程同时读,并且保证不会读到过期的值,但是只能被单线程写(有一种情况可以被多线程写,就是写入的值不依赖于原值),在get操作里只需要读不需要写共享变量count和value,所以可以不用加锁。之所以不会读到过期的值,是根据java内存模型的happen before原则,对volatile字段的写入操作先于读操作,即使两个线程同时修改和获取volatile变量,get操作也能拿到最新的值,这是用volatile替换锁的经典应用场景;

接下来看看Put操作:

V put(K key, int hash, V value, boolean onlyIfAbsent) {  
     lock();  
     try {  
         int c = count;  
         if (c++ > threshold) // ensure capacity  
             rehash();  
         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;  
         if (e != null) {  
             oldValue = e.value;  
             if (!onlyIfAbsent)  
                 e.value = value;  
         }  
         else {  
             oldValue = null;  
             ++modCount;  
             tab[index] = new HashEntry<K,V>(key, hash, first, value);  
             count = c; // write-volatile  
         }  
         return oldValue;  
     } finally {  
         unlock();  
     }  
 }

先上个锁,做个预判,容量+1的时候需不需要扩容,这个判断相比于HashMap做的非常精妙,因为HashMap是在增加元素之后才判断是否需要扩容,如果我扩容之后不增加元素,就白扩容了;

通过索引定位到HashEntry,判断如果不为空,替换节点的value,否则new一个Entry,将此Entry插入到链头,由构造函数得知,它的next则是first;

扩容默认也是会创建两倍容量的数组,进行计算将原数组的数据插入到新数组,ConcurrentHashMap不会对整个容器进行扩容,而只对某个segment进行扩容;

 

ConcurrentHashMap的size方法

需要统计所有segment的元素总和,为了不影响性能,它会先2次尝试不锁住segment来统计大小,如果统计的过程中count发生了变化,则会把所有的segment的put和remove方法全部锁住;通过modCount来协助判断容器是否发生变化;统计size前后判断modCount是否发生变化即可;

参考:http://www.cnblogs.com/ITtangtang/p/3948786.html

 

标签:

版权申明:本站文章部分自网络,如有侵权,请联系:west999com@outlook.com
特别注意:本站所有转载文章言论不代表本站观点,本站所提供的摄影照片,插画,设计作品,如需使用,请与原作者联系,版权归原作者所有

上一篇:设计模式之装饰器模式 Decorator 包装模式 wrapper 优点 缺点

下一篇:并发执行任务