ConcurrentHashMap的锁分离技术

 

 

concurrenthashmap是一个非常好的map实现，在高并发操作的场景下会有非常好的效率。实现的目的主要是为了避免同步操作时对整个map对象进行锁定从而提高并发访问能力。

 

ConcurrentHashMap 类中包含两个静态内部类 HashEntry 和 Segment。HashEntry 用来封装映射表的键 / 值对；Segment 用来充当锁的角色，每个 Segment 对象守护整个散列映射表的若干个桶。每个桶是由若干个 HashEntry 对象链接起来的链表。一个 ConcurrentHashMap 实例中包含由若干个 Segment 对象组成的数组。

 

 

 static final class HashEntry<K,V> { 
        final K key;                       // 声明 key 为 final 型
        final int hash;                   // 声明 hash 值为 final 型 
        volatile V value;                 // 声明 value 为 volatile 型
        final HashEntry<K,V> next;      // 声明 next 为 final 型 

        HashEntry(K key, int hash, HashEntry<K,V> next, V value) { 
            this.key = key; 
            this.hash = hash; 
            this.next = next; 
            this.value = value; 
        } 
 } 

 

 

static final class Segment<K,V> extends ReentrantLock implements Serializable { 

        transient volatile int count;  //在本 segment 范围内，包含的 HashEntry 元素的个数
				                    //volatile 型

        transient int modCount;     //table 被更新的次数

        transient int threshold;    //默认容量

	final float loadFactor;    //装载因子

        /** 
         * table 是由 HashEntry 对象组成的数组
         * 如果散列时发生碰撞，碰撞的 HashEntry 对象就以链表的形式链接成一个链表
         * table 数组的数组成员代表散列映射表的一个桶        
         */ 
        transient volatile HashEntry<K,V>[] table; 
      

        /** 
         * 根据 key 的散列值，找到 table 中对应的那个桶（table 数组的某个数组成员）
	 *     把散列值与 table 数组长度减 1 的值相“与”，得到散列值对应的 table 数组的下标
         *     然后返回 table 数组中此下标对应的 HashEntry 元素
	 * 即这个段中链表的第一个元素
         */ 
        HashEntry<K,V> getFirst(int hash) { 
            HashEntry<K,V>[] tab = table;             
            return tab[hash & (tab.length - 1)]; 
        } 



        Segment(int initialCapacity, float lf) { 
            loadFactor = lf; 
            setTable(HashEntry.<K,V>newArray(initialCapacity)); 
        } 

        /** 
         * 设置 table 引用到这个新生成的 HashEntry 数组
         * 只能在持有锁或构造函数中调用本方法
         */ 
        void setTable(HashEntry<K,V>[] newTable) {             
            threshold = (int)(newTable.length * loadFactor); 
            table = newTable; 
        }        
 } 

 注意Segment继承了ReentrantLock 锁

 

左边便是Hashtable的实现方式---锁整个hash表；而右边则是ConcurrentHashMap的实现方式---锁桶（或段）。 ConcurrentHashMap将hash表分为16个桶（默认值），诸如get,put,remove等常用操作只锁当前需要用到的桶。试想，原来 只能一个线程进入，现在却能同时16个写线程进入（写线程才需要锁定，而读线程几乎不受限制，之后会提到），并发性的提升是显而易见的。

    更令人惊讶的是ConcurrentHashMap的读取并发，因为在读取的大多数时候都没有用到锁定，所以读取操作几乎是完全的并发操作，而写操作锁定的粒度又非常细，比起之前又更加快速（这一点在桶更多时表现得更明显些）。只有在求size等操作时才需要锁定整个表。而在迭代时，ConcurrentHashMap使用了不同于传统集合的快速失败迭代器（见之前的文章《JAVA API备忘---集合》）的另一种迭代方式，我们称为弱一致迭代器。在这种迭代方式中，当iterator被创建后集合再发生改变就不再是抛出ConcurrentModificationException，取而代之的是在改变时new新的数据从而不影响原有的数据，iterator完成后再将头指针替换为新的数据，这样iterator线程可以使用原来老的数据，而写线程也可以并发的完成改变，更重要的，这保证了多个线程并发执行的连续性和扩展性，是性能提升的关键。

    接下来，让我们看看ConcurrentHashMap中的几个重要方法，心里知道了实现机制后，使用起来就更加有底气。

 

    ConcurrentHashMap中主要实体类就是三个：ConcurrentHashMap（整个Hash表）,Segment（桶），HashEntry（节点），对应上面的图可以看出之间的关系。

    get方法（请注意，这里分析的方法都是针对桶的，因为ConcurrentHashMap的最大改进就是将粒度细化到了桶上），首先判断了当前桶的数据 个数是否为0，为0自然不可能get到什么，只有返回null，这样做避免了不必要的搜索，也用最小的代价避免出错。然后得到头节点（方法将在下面涉及） 之后就是根据hash和key逐个判断是否是指定的值，如果是并且值非空就说明找到了，直接返回；程序非常简单，但有一个令人困惑的地方，这句 return readValueUnderLock(e)到底是用来干什么的呢？研究它的代码，在锁定之后返回一个值。但这里已经有一句V v = e.value得到了节点的值，这句return readValueUnderLock(e)是否多此一举？事实上，这里完全是为了并发考虑的，这里当v为空时，可能是一个线程正在改变节点，而之前的get操作都未进行锁定，根据bernstein条件，读后写或写后读都会引起数据的不一致，所以这里要对这个e重新上锁再读一遍，以保证得到的是正确值，这里不得不佩服Doug Lee思维的严密性。整个get操作只有很少的情况会锁定，相对于之前的Hashtable，并发是不可避免的啊！

 

get操作不需要锁。第一步是访问count变量，这是一个volatile变量，由于所有的修改操作在进行结构修改时都会在最后一步写count变量，通过这种机制保证get操作能够得到几乎最新的结构更新。对于非结构更新，也就是结点值的改变，由于HashEntry的value变量是volatile的，也能保证读取到最新的值。接下来就是对hash链进行遍历找到要获取的结点，如果没有找到，直接访回null。对hash链进行遍历不需要加锁的原因在于链指针next是final的。但是头指针却不是final的，这是通过getFirst(hash)方法返回，也就是存在table数组中的值。这使得getFirst(hash)可能返回过时的头结点，例如，当执行get方法时，刚执行完getFirst(hash)之后，另一个线程执行了删除操作并更新头结点，这就导致get方法中返回的头结点不是最新的。这是可以允许，通过对count变量的协调机制，get能读取到几乎最新的数据，虽然可能不是最新的。要得到最新的数据，只有采用完全的同步。

V get(Object key, int hash) {
    if (count != 0) { // read-volatile
        HashEntry 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 e) {
    lock();
    try {
        return e.value;
    } finally {
        unlock();
    }
}

 

put操作一上来就锁定了整个segment，这当然是为了并发的安全，修改数据是不能并发进行的，必须得有个判断是否超限的语句以确保容量不足时能够rehash，而比较难懂的是这句int index = hash & (tab.length - 1)，原来segment里面才是真正的hashtable，即每个segment是一个传统意义上的hashtable,如上图，从两者的结构就可以看出区别，这里就是找出需要的entry在table的哪一个位置，之后得到的entry就是这个链的第一个节点，如果e!=null，说明找到了，这是就要替换节点的值（onlyIfAbsent == false），否则，我们需要new一个entry，它的后继是first，而让tab[index]指向它，什么意思呢？实际上就是将这个新entry插入到链头，剩下的就非常容易理解了。

 

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

  

   remove操作非常类似put，但要注意一点区别，中间那个for循环是做什么用的呢？（*号标记）从代码来看，就是将定位之后的所有entry克隆并拼回前面去，但有必要吗？每次删除一个元素就要将那之前的元素克隆一遍？这点其实是由entry 的不变性来决定的，仔细观察entry定义，发现除了value，其他所有属性都是用final来修饰的，这意味着在第一次设置了next域之后便不能再 改变它，取而代之的是将它之前的节点全都克隆一次。至于entry为什么要设置为不变性，这跟不变性的访问不需要同步从而节省时间有关，关于不变性的更多 内容，请参阅之前的文章《线程高级---线程的一些编程技巧》

 

V remove(Object key, int hash, Object value) {
    lock();
    try {
        int c = count - 1;
        HashEntry[] tab = table;
        int index = hash & (tab.length - 1);
        HashEntry first = (HashEntry)tab[index];
        HashEntry 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 newFirst = e.next;
            *    for (HashEntry p = first; p != e; p = p.next)
            *        newFirst = new HashEntry(p.key, p.hash,
                                                  newFirst, p.value);
                tab[index] = newFirst;
                count = c; // write-volatile
            }
        }
        return oldValue;
    } finally {
        unlock();
    }
}

 

 

探索 ConcurrentHashMap 高并发性的实现机制：
http://www.ibm.com/developerworks/cn/java/java-lo-concurrenthashmap/

 

 

 

ConcurrentHashMap之实现细节
http://www.iteye.com/topic/344876

 

Map的并发处理（ConcurrentHashMap）

http://zl198751.iteye.com/blog/907927

 

集合框架 Map篇（4）----ConcurrentHashMap

http://hi.baidu.com/yao1111yao/blog/item/232f2dfc55fbcd5ad7887d9f.html

 

 

java ConcurrentHashMap中的一点点迷惑

http://icanfly.iteye.com/blog/1450165