Java并发编程：并发容器之ConcurrentHashMap（转载）

时间 2019-11-25

标签 java 并发编程容器 concurrenthashmap 转载栏目 Java 繁體版

原文原文链接

下面这部份内容转载自：java

　　http://www.haogongju.net/art/2350374node

　　JDK5中添加了新的concurrent包，相对同步容器而言，并发容器经过一些机制改进了并发性能。由于同步容器将全部对容器状态的访问都c++

串行化了，这样保证了线程的安全性，因此这种方法的代价就是严重下降了并发性，当多个线程竞争容器时，吞吐量严重下降。所以Java5.0开数组

始针对多线程并发访问设计，提供了并发性能较好的并发容器，引入了java.util.concurrent包。与Vector和Hashtable、安全

Collections.synchronizedXxx()同步容器等相比，util.concurrent中引入的并发容器主要解决了两个问题：
1）根据具体场景进行设计，尽可能避免synchronized，提供并发性。
2）定义了一些并发安全的复合操做，而且保证并发环境下的迭代操做不会出错。数据结构

　　util.concurrent中容器在迭代时，能够不封装在synchronized中，能够保证不抛异常，可是未必每次看到的都是"最新的、当前的"数据。多线程

　　下面是对并发容器的简单介绍：并发

　　ConcurrentHashMap代替同步的Map（Collections.synchronized（new HashMap()）），众所周知，HashMap是根据散列值分段存储的，同步Map在同步的时候锁住了全部的段，而ConcurrentHashMap加锁的时候根据散列值锁住了散列值锁对应的那段，所以提升了并发性能。ConcurrentHashMap也增长了对经常使用复合操做的支持，好比"若没有则添加"：putIfAbsent()，替换：replace()。这2个操做都是原子操做。app

　　CopyOnWriteArrayList和CopyOnWriteArraySet分别代替List和Set，主要是在遍历操做为主的状况下来代替同步的List和同步的Set，这也就是上面所述的思路：迭代过程要保证不出错，除了加锁，另一种方法就是"克隆"容器对象。ssh

　　ConcurrentLinkedQuerue是一个先进先出的队列。它是非阻塞队列。

　 ConcurrentSkipListMap能够在高效并发中替代SoredMap（例如用Collections.synchronzedMap包装的TreeMap）。

　　ConcurrentSkipListSet能够在高效并发中替代SoredSet（例如用Collections.synchronzedSet包装的TreeMap）。

　　本篇文章着重讲解2个并发容器：ConcurrentHashMap和CopyOnWriteArrayList其中的ConcurrentHashMap，CopyOnWriteArrayList在下一篇文章中讲述。

　　原文连接：http://www.iteye.com/topic/1103980

　　你们都知道HashMap是非线程安全的，Hashtable是线程安全的，可是因为Hashtable是采用synchronized进行同步，至关于全部线程进行读写时都去竞争一把锁，致使效率很是低下。

　　ConcurrentHashMap能够作到读取数据不加锁，而且其内部的结构可让其在进行写操做的时候可以将锁的粒度保持地尽可能地小，不用对整个ConcurrentHashMap加锁。

ConcurrentHashMap的内部结构

　　ConcurrentHashMap为了提升自己的并发能力，在内部采用了一个叫作Segment的结构，一个Segment其实就是一个类Hash Table的结构，Segment内部维护了一个链表数组，咱们用下面这一幅图来看下ConcurrentHashMap的内部结构：

　　从上面的结构咱们能够了解到，ConcurrentHashMap定位一个元素的过程须要进行两次Hash操做，第一次Hash定位到Segment，第二次Hash定位到元素所在的链表的头部，所以，这一种结构的带来的反作用是Hash的过程要比普通的HashMap要长，可是带来的好处是写操做的时候能够只对元素所在的Segment进行加锁便可，不会影响到其余的Segment，这样，在最理想的状况下，ConcurrentHashMap能够最高同时支持Segment数量大小的写操做（恰好这些写操做都很是平均地分布在全部的Segment上），因此，经过这一种结构，ConcurrentHashMap的并发能力能够大大的提升。

Segment

　　咱们再来具体了解一下Segment的数据结构：

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

transient volatile int count;

transient int modCount;

transient int threshold;

transient volatile HashEntry<K,V>[] table;

final float loadFactor;

}

详细解释一下Segment里面的成员变量的意义：

count：Segment中元素的数量

modCount：对table的大小形成影响的操做的数量（好比put或者remove操做）
threshold：阈值，Segment里面元素的数量超过这个值依旧就会对Segment进行扩容
table：链表数组，数组中的每个元素表明了一个链表的头部
loadFactor：负载因子，用于肯定threshold

HashEntry

　　Segment中的元素是以HashEntry的形式存放在链表数组中的，看一下HashEntry的结构：

static final class HashEntry<K,V> {

final K key;

final int hash;

volatile V value;

final HashEntry<K,V> next;

}

能够看到HashEntry的一个特色，除了value之外，其余的几个变量都是final的，这样作是为了防止链表结构被破坏，出现ConcurrentModification的状况。

ConcurrentHashMap的初始化

　　下面咱们来结合源代码来具体分析一下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;
    }
    segmentShift = 32 - sshift;
    segmentMask = ssize - 1;
    this.segments = Segment.newArray(ssize);
  
    if (initialCapacity > MAXIMUM_CAPACITY)
        initialCapacity = MAXIMUM_CAPACITY;
    int c = initialCapacity / ssize;
    if (c * ssize < initialCapacity)
        ++c;
    int cap = 1;
    while (cap < c)
        cap <<= 1;
  
    for (int i = 0; i < this.segments.length; ++i)
        this.segments[i] = new Segment<K,V>(cap, loadFactor);
}

CurrentHashMap的初始化一共有三个参数，一个initialCapacity，表示初始的容量，一个loadFactor，表示负载参数，最后一个是concurrentLevel，表明ConcurrentHashMap内部的Segment的数量，ConcurrentLevel一经指定，不可改变，后续若是ConcurrentHashMap的元素数量增长致使ConrruentHashMap须要扩容，ConcurrentHashMap不会增长Segment的数量，而只会增长Segment中链表数组的容量大小，这样的好处是扩容过程不须要对整个ConcurrentHashMap作rehash，而只须要对Segment里面的元素作一次rehash就能够了。

　　整个ConcurrentHashMap的初始化方法仍是很是简单的，先是根据concurrentLevel来new出Segment，这里Segment的数量是不大于concurrentLevel的最大的2的指数，就是说Segment的数量永远是2的指数个，这样的好处是方便采用移位操做来进行hash，加快hash的过程。接下来就是根据intialCapacity肯定Segment的容量的大小，每个Segment的容量大小也是2的指数，一样使为了加快hash的过程。

　　这边须要特别注意一下两个变量，分别是segmentShift和segmentMask，这两个变量在后面将会起到很大的做用，假设构造函数肯定了Segment的数量是2的n次方，那么segmentShift就等于32减去n，而segmentMask就等于2的n次方减一。

ConcurrentHashMap的get操做

　　前面提到过ConcurrentHashMap的get操做是不用加锁的，咱们这里看一下其实现：

public V get(Object key) {
    int hash = hash(key.hashCode());
    return segmentFor(hash).get(key, hash);
}

看第三行，segmentFor这个函数用于肯定操做应该在哪个segment中进行，几乎对ConcurrentHashMap的全部操做都须要用到这个函数，咱们看下这个函数的实现：

final Segment<K,V> segmentFor(int hash) {
    return segments[(hash >>> segmentShift) & segmentMask];
}

这个函数用了位操做来肯定Segment，根据传入的hash值向右无符号右移segmentShift位，而后和segmentMask进行与操做，结合咱们以前说的segmentShift和segmentMask的值，就能够得出如下结论：假设Segment的数量是2的n次方，根据元素的hash值的高n位就能够肯定元素到底在哪个Segment中。

　　在肯定了须要在哪个segment中进行操做之后，接下来的事情就是调用对应的Segment的get方法：

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

先看第二行代码，这里对count进行了一次判断，其中count表示Segment中元素的数量，咱们能够来看一下count的定义：

transient volatile int count;

能够看到count是volatile的，实际上这里里面利用了volatile的语义：

　　对volatile字段的写入操做happens-before于每个后续的同一个字段的读操做。

　　由于实际上put、remove等操做也会更新count的值，因此当竞争发生的时候，volatile的语义能够保证写操做在读操做以前，也就保证了写操做对后续的读操做都是可见的，这样后面get的后续操做就能够拿到完整的元素内容。而后，在第三行，调用了getFirst()来取得链表的头部：

HashEntry<K,V> getFirst(int hash) {
    HashEntry<K,V>[] tab = table;
    return tab[hash & (tab.length - 1)];
}

一样，这里也是用位操做来肯定链表的头部，hash值和HashTable的长度减一作与操做，最后的结果就是hash值的低n位，其中n是HashTable的长度以2为底的结果。

　　在肯定了链表的头部之后，就能够对整个链表进行遍历，看第4行，取出key对应的value的值，若是拿出的value的值是null，则可能这个key，value对正在put的过程当中，若是出现这种状况，那么就加锁来保证取出的value是完整的，若是不是null，则直接返回value。

ConcurrentHashMap的put操做

　　看完了get操做，再看下put操做，put操做的前面也是肯定Segment的过程，这里再也不赘述，直接看关键的segment的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();
    }
}

首先对Segment的put操做是加锁完成的，而后在第五行，若是Segment中元素的数量超过了阈值（由构造函数中的loadFactor算出）这须要进行对Segment扩容，而且要进行rehash，关于rehash的过程你们能够本身去了解，这里不详细讲了。

　　第8和第9行的操做就是getFirst的过程，肯定链表头部的位置。

　　第11行这里的这个while循环是在链表中寻找和要put的元素相同key的元素，若是找到，就直接更新更新key的value，若是没有找到，则进入21行这里，生成一个新的HashEntry而且把它加到整个Segment的头部，而后再更新count的值。

ConcurrentHashMap的remove操做

　　Remove操做的前面一部分和前面的get和put操做同样，都是定位Segment的过程，而后再调用Segment的remove方法：

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

首先remove操做也是肯定须要删除的元素的位置，不过这里删除元素的方法不是简单地把待删除元素的前面的一个元素的next指向后面一个就完事了，咱们以前已经说过HashEntry中的next是final的，一经赋值之后就不可修改，在定位到待删除元素的位置之后，程序就将待删除元素前面的那一些元素所有复制一遍，而后再一个一个从新接到链表上去，看一下下面这一幅图来了解这个过程：

　　假设链表中原来的元素如上图所示，如今要删除元素3，那么删除元素3之后的链表就以下图所示：

ConcurrentHashMap的size操做

　　在前面的章节中，咱们涉及到的操做都是在单个Segment中进行的，可是ConcurrentHashMap有一些操做是在多个Segment中进行，好比size操做，ConcurrentHashMap的size操做也采用了一种比较巧的方式，来尽可能避免对全部的Segment都加锁。

　　前面咱们提到了一个Segment中的有一个modCount变量，表明的是对Segment中元素的数量形成影响的操做的次数，这个值只增不减，size操做就是遍历了两次Segment，每次记录Segment的modCount值，而后将两次的modCount进行比较，若是相同，则表示期间没有发生过写入操做，就将原先遍历的结果返回，若是不相同，则把这个过程再重复作一次，若是再不相同，则就须要将全部的Segment都锁住，而后一个一个遍历了，具体的实现你们能够看ConcurrentHashMap的源码，这里就不贴了。

　　另外2篇讲述关于ConcurrentHashMap原理的两篇文章：

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

　　《聊聊并发（四）深刻分析ConcurrentHashMap》：http://ifeve.com/ConcurrentHashMap/