`
frank-liu
  • 浏览: 1682362 次
  • 性别: Icon_minigender_1
  • 来自: 北京
社区版块
存档分类
最新评论

java concurrency: ConcurrentHashMap

 
阅读更多

引言

    以前有几次碰到过一个有意思的多线程问题,当时的场景看起来比较简单。有两个线程,他们都需要写数据到统一的一个数据结构里,因为这两个线程是相互独立的,在他们执行的过程中我们将每个线程的一组名值对输出。当时觉得这不过是一个很简单的问题,我们可以用一个HashTable或者HashMap就可以了。我们可以保证两个线程访问数据的key是不同的,看起来应该不会存在多线程访问的冲突。实际运行的时候却发现有的时候会出现一个线程执行结束之后另外一个却被卡在那里的现象。后来看了一些相关的材料后才知道,问题似乎不是那么简单。

    关于HashMap在并行情况下会出现的问题,这篇文章有一个很详细的分析。

ConcurrenctHashMap实现

    从官方的文档可以看到,在多线程的情况下推荐我们使用ConcurrentHashMap。那么和HashMap比起来,它有什么特殊的地方呢?这里,我们针对不同的各方面进行一下详细的分析。 

结构

整体结构

    笼统的来说,ConcurrentHashMap里面也是由一个个的HashEntry组成的。这里HashEntry的定义如下:

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

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

//... omitted
    }

    从代码中可以看到,这里是一个很典型的链表的节点。和前面的HashMap有点不同的地方在于,这里的value和next都采用了volatile的修饰。这样保证在有多个线程访问的情况下不会出现数据不一致性。(关于volatile相关的分析可以见我的这篇文章。)而从我们了解到HashMap的结构来说,一个HashEntry这样的数组应该就组成了ConcurrentHashMap的整体结构。这样来理解不完全准确。因为在ConcurrentHashMap里,考虑到它需要经历多线程的访问,它的结构是分段的。

    如何理解这个分段呢?我们来看一些代码。在ConcurrentHashMap里有这么一个声明:

final Segment<K,V>[] segments;

    而Segment的大体定义如下:

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

        private static final long serialVersionUID = 2249069246763182397L;

        static final int MAX_SCAN_RETRIES =
            Runtime.getRuntime().availableProcessors() > 1 ? 64 : 1;

        /**
         * The per-segment table. Elements are accessed via
         * entryAt/setEntryAt providing volatile semantics.
         */
        transient volatile HashEntry<K,V>[] table;

        transient int count;

        transient int modCount;

        transient int threshold;

        final float loadFactor;

        Segment(float lf, int threshold, HashEntry<K,V>[] tab) {
            this.loadFactor = lf;
            this.threshold = threshold;
            this.table = tab;
        }
}

    这里有一个典型的定义就是里面定义了HashEntry数组。为什么要将整个ConcurrentHashMap拆成Segment的数组呢?实际上,每个Segment我们可以将其看成是一个HashMap,处于多线程情况的考虑,我们分出了多个Segment,每个Segment对应一个访问的线程,这样在有并发访问的情况下,只要两个线程的数据访问操作不是在同一个Segment里面,他们是可以并行执行的。这里体现出来了一个ConcurrentHashMap设计的重要思路,就是分离锁。

    综上所述,ConcurrentHashMap的结构则如下图所示:

 

Segment详解

    Segment里面的设计看起来和一个普通的HashMap差别不大。如果仔细去分析里面的方法的话,会发现很多有意思的地方。这里就列举一下里面的方法:

方法名 说明
put 添加元素
rehash 调整表长度并拷贝元素到新表
remove 删除元素
replace 替换元素
clear 清除元素
scanAndLockForPut 根据给定的key找到特定的节点来添加元素,并加锁
scanAndLock 根据给定的key找到特定的节点,并加锁

    似乎很奇怪,这里所有的访问方法都是和写相关的,并没有定义和读相关的。因为在ConcurrentHashMap里,对元素的读操作是可以并行执行的,他们之间不会产生多线程的问题。所以在这里我们不需要做特殊的处理,只要通过定义找到该元素并返回就可以了。而这里的写操作则不一样,因为牵涉到对元素的修改,如果我们要保证该部分的元素是线程安全的,我们就需要加锁处理了。这些写操作里都有加锁和释放锁的行为。这里我们针对put方法和rehash方法的实现细节详细分析一下。put方法的代码如下:

final V put(K key, int hash, V value, boolean onlyIfAbsent) {
	// 获取锁以及插入的node元素
        HashEntry<K,V> node = tryLock() ? null :
		scanAndLockForPut(key, hash, value);
	V oldValue;
	try {
	    HashEntry<K,V>[] tab = table;
	    int index = (tab.length - 1) & hash;
            // 通过entryAt获取到需要加入值的slot
 	    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 { //如果在slot里没有找到匹配的key
            	    if (node != null) //如果在前面的scanAndLockForPut方法里成功创建了node节点对象
                        node.setNext(first);
                    else
                        node = new HashEntry<K,V>(hash, key, value, first);
                    int c = count + 1;
                    if (c > threshold && tab.length < MAXIMUM_CAPACITY)
                    // 判断是否达到表长度的阀值,如果是则要调整表长度
                        rehash(node);
                    else
                        setEntryAt(tab, index, node);
                    ++modCount;
                    count = c;
                    oldValue = null;
                    break;
                }
            }
        } finally {
    	    unlock();
        }
    return oldValue;
}

     这里的代码其实并不是太复杂。前面的scanAndLockForPut方法已经获得Segment部分的锁。所以在后面的finally部分需要调用unlock方法来释放锁。除了这一部分,我们再想想put操作需要注意的几个点,一个就是如果需要放置的元素已经有值了,我们则需要更新这个旧的值。而如果这个值不存在,我们可能需要新建一个元素添加进去,这里还要判断的一个点就是当长度达到一个阀值的时候会调用rehash方法来调整表的长度。

    我们再来看看rehash方法的具体操作:

private void rehash(HashEntry<K,V> node) {
            HashEntry<K,V>[] oldTable = table;
            int oldCapacity = oldTable.length;
            int newCapacity = oldCapacity << 1; //新表长度加倍,保证长度一直是2的幂
            threshold = (int)(newCapacity * loadFactor);
            HashEntry<K,V>[] newTable =
                (HashEntry<K,V>[]) new HashEntry[newCapacity];
            int sizeMask = newCapacity - 1; // sizeMask用来通过和hash值与计算求下标
            for (int i = 0; i < oldCapacity ; i++) {
                HashEntry<K,V> e = oldTable[i];
                if (e != null) {
                    HashEntry<K,V> next = e.next;
                    int idx = e.hash & sizeMask;
                    if (next == null)   //  Single node on list
                        newTable[idx] = e;
                    else { // Reuse consecutive sequence at same slot
                        HashEntry<K,V> lastRun = e;
                        int lastIdx = idx;
                        for (HashEntry<K,V> last = next;
                             last != null;
                             last = last.next) {
                            int k = last.hash & sizeMask;
                            if (k != lastIdx) {
                                lastIdx = k;
                                lastRun = last;
                            }
                        }
                        newTable[lastIdx] = lastRun;
                        // Clone remaining nodes
                        for (HashEntry<K,V> p = e; p != lastRun; p = p.next) {
                            V v = p.value;
                            int h = p.hash;
                            int k = h & sizeMask;
                            HashEntry<K,V> n = newTable[k];
                            newTable[k] = new HashEntry<K,V>(h, p.key, v, n);
                        }
                    }
                }
            }
            int nodeIndex = node.hash & sizeMask; // add the new node
            node.setNext(newTable[nodeIndex]);
            newTable[nodeIndex] = node;
            table = newTable;
        }

    前面代码里有一个部分比较让人困惑,就是在当我们找到一个next不为null的节点的时候我们遍历这个链表,并记录lastRun, lastIdx。因为我们这里重新计算的k是通过每个元素的hash值和新的sizeMask做与元算得到的。这里原来在一个slot里的元素可能不一定映射到同一个下面了,所以这里就保留了最后一个指向相同索引的元素,然后将这个引用直接赋值给新表中的元素。这样就相当于将链表的一截截取到一个新的slot中。这部分的代码其实是想取一点点巧,这样取得到最后一段index相同的元素,可以直接搬过去就可以了,而不用像后面的循环里每次重新构造一个同样的对象再加入到链表的头部。

   后面接着的for循环将原来旧表中的内容不断拷贝到新的slot中。

   后面4行代码则比较简单,只是将新的节点加入到指定的slot中。这里是相当于将节点加入到链表的头部。

   总的来说,这部分的代码看起来很多,里面却没有用到多少技巧。简单的来说还是创建一个新的数组,再将原有的元素重新映射。因为有的元素是一个链表,所以要遍历里面所有链表的内容。

    有了这部分的分析,其他部分的理解也就很容易了。不再赘述。

构造函数

    有了前面对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;
        }
        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;
    }

   这里的代码看起来也不少,实际分析起来也不难,我们一步步看过来。这里有一个关键的参数就是concurrencyLevel,它表示我们希望提供的并行度,也就是说我们希望结构里可以支持多少个并行的线程。假设我们这里设定的concurrencyLevel = 16,也就是说我们希望能够同时支持16个线程访问。那么从前面讨论分离锁的思路来说,我们就应该要将整个表分成16个段。这里我们将看到,当while (ssize < concurrencyLevel)这个循环部分执行完毕之后,ssize = 16。

    而后面计算c的部分则用来确认一个Segment里面的数组大致长度。既然定的是16个线程访问那么它的长度应该就是>=initialCapacity / ssize。当然,这里还只是一个粗略的估计,因为在这里还有一个和HashMap里相同的地方,就是它还必须是2的指数。所以后面的while (cap < c)循环里通过将cap不断长度翻倍来达到这个目标。而cap最初始的值就是2。

    后面定义的Segment[]数组的长度肯定就是16了,表示有16个Segment。这里并没有一次把所有的Segment段都创建了,只是创建了一个。

典型方法分析

    我们先来看看get方法的具体实现:

public V get(Object key) {
        Segment<K,V> s; // manually integrate access methods to reduce overhead
        HashEntry<K,V>[] tab;
        int h = hash(key);
        long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
        if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null &&
            (tab = s.table) != null) {
            for (HashEntry<K,V> e = (HashEntry<K,V>) UNSAFE.getObjectVolatile
                     (tab, ((long)(((tab.length - 1) & h)) << TSHIFT) + TBASE);
                 e != null; e = e.next) {
                K k;
                if ((k = e.key) == key || (e.hash == h && key.equals(k)))
                    return e.value;
            }
        }
        return null;
    }

    前面这部分代码最麻烦的地方在于根据key计算出来hash值之后映射到特定segment以及segment里面的index。long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;这一部分是用来计算所在segment的索引。而((tab.length - 1) & h)) << TSHIFT) + TBASE则用来计算映射到特定HashEntry的index。

    我们再看看remove方法:

public V remove(Object key) {
        int hash = hash(key);
        Segment<K,V> s = segmentForHash(hash);
        return s == null ? null : s.remove(key, hash, null);
    }

     这里用另外一个方法segmentForHash可以根据hash值得到所处的segment段。然后再调用segment对象里的remove方法。segment里面的remove方法分析思路和前面的put方法类似,都是通过加锁然后操作,这里就不详细分析。有兴趣的可以去对照参考一下。

    前面的segmentForHash方法的实现如下:

private Segment<K,V> segmentForHash(int h) {
        long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
        return (Segment<K,V>) UNSAFE.getObjectVolatile(segments, u);
    }

    这里和前面get方法里部分的代码一样,关键还是值的映射。这里的难点就在于怎么通过移位比较来计算出所在的segment段以及映射到的索引。

总结

    关于ConcurrentHashMap的分析这里主要是参照了openjdk的一个实现。不同实现版本的细节存在一些差别。不过他们的整体思想都是一样的,就是锁分离的机制。将一个大的HashMap拆分成多个segment,然后在每个segment段运用独立的锁来操作。这样多个线程可以并发的操作不同segment。当然这种机制带来的一个复杂的地方就是根据一个hash值映射到某个segment及某个特定的index。另外因为可能有多个线程访问同一个段,所以里面关键的数据值是声明为volatile的,这是为了保证一个线程修改后的结果能够马上被后续访问的线程看到。总的来说,ConcurrentHashMap带来了一定程度的可定制化并行度。

参考材料

openjdk

Java concurrency in practice

http://www.ibm.com/developerworks/cn/java/java-lo-concurrenthashmap/

http://coolshell.cn/articles/9606.html

http://marlonyao.iteye.com/blog/344876

  • 大小: 21.4 KB
分享到:
评论

相关推荐

    Java并发编程实践(java concurrency in practice)pdf (java多线程总结.ppt)

    6. **并发集合**:Java提供了线程安全的集合,如`ConcurrentHashMap`、`CopyOnWriteArrayList`等,这些集合在并发环境中性能优异。书中会分析它们的设计原理和使用场景。 7. **线程池**:`ExecutorService`是线程池...

    Java Concurrency in Practice.zip

    《Java Concurrency in Practice》是Java并发编程领域的一本经典著作,由Brian Goetz、Tim Peierls、Joshua Bloch、Joseph Bowles和Doug Lea等专家共同编写。这本书深入探讨了Java平台上的多线程和并发编程,旨在...

    Java concurrency集合之ConcurrentHashMap_动力节点Java学院整理

    【Java并发集合之ConcurrentHashMap详解】 在Java的并发编程中,`ConcurrentHashMap`是一个极其重要的工具,它是线程安全的哈希映射表,提供了高效且安全的并发访问性能。相较于传统的线程安全的`Hashtable`和非...

    concurrency:Java Concurrency in Practice源代码-java source code

    《Java并发编程实践》(Java Concurrency in Practice)是一本深度探讨Java多线程和并发编程的经典著作。这本书由Brian Goetz、Tim Peierls、Joshua Bloch、David Holmes和Doug Lea合著,旨在帮助Java开发者理解和解决...

    Java-Concurrency:Java并发学习演示

    6. **并发集合**:Java并发库提供了一些线程安全的集合,如`ConcurrentHashMap`, `CopyOnWriteArrayList`, `ConcurrentLinkedQueue`等,它们内部实现了线程安全的算法,允许在并发环境下高效操作。 7. **Executor...

    JavaConcurrency:这是一个用于Java并发包的学习程序

    Java并发是Java编程中至关重要的一部分,它涉及到多线程、同步机制、线程池和并发集合等关键概念。Java并发包(java.util.concurrent)提供了一系列的类和接口,旨在简化多线程环境下的编程,提高应用程序的性能和可...

    java-concurrency:java并发教程

    此外,`java.util.concurrent`包提供了许多线程安全的数据结构,如`ConcurrentHashMap`,它们在内部实现了高级同步策略,以提供高效且线程安全的并发操作。 在这个"java-concurrency-master"压缩包中,很可能是包含...

    Java Lesson: Concurrency

    线程安全的类,如`ConcurrentHashMap`、`AtomicInteger`等,也是Java并发编程的重要组成部分。这些类提供了内在的并发控制,使得在多线程环境下操作它们更加安全和高效。 理解Java并发编程不仅包括掌握各种API的...

    Java Concurrency in Practice中文版

    《Java Concurrency in Practice》是Java并发编程领域的一本经典著作,由Brian Goetz、Tim Peierls、Joshua Bloch、Joe Bowbeer、David Holmes和Doug Lea合著,国内有热心人士进行了中文翻译,使得更多的中国开发者...

    Java Concurrency In Practice Learning Note

    本笔记将深入探讨《Java Concurrency In Practice》这本书中的核心概念,结合Guava库的实际使用案例,帮助读者理解并掌握Java并发编程的精髓。 首先,我们来了解Java并发的基础知识。Java提供了丰富的并发工具类,...

    Concurrent_Programming+Java Concurrency in Practice+langspec

    首先,"Java Concurrency in Practice"是Java并发编程的经典之作,由Brian Goetz、Tim Peierls、Joshua Bloch、David Holmes和Doug Lea合著。这本书提供了一套实用的指导原则、设计模式和最佳实践,帮助Java开发者...

    Java并发系列之ConcurrentHashMap源码分析

    例如DEFAULT_INITIAL_CAPACITY、DEFAULT_LOAD_FACTOR、DEFAULT_CONCURRENCY_LEVEL、MAXIMUM_CAPACITY、MIN_SEGMENT_TABLE_CAPACITY、MAX_SEGMENTS、RETRIES_BEFORE_LOCK等,它们控制着ConcurrentHashMap的初始化和...

    java-concurrency:代码审查清单

    2. **并发容器**:Java提供了许多线程安全的容器,如`ConcurrentHashMap`、`CopyOnWriteArrayList`和`BlockingQueue`等,它们设计用于在多线程环境中高效地操作数据。理解这些容器的特性和使用场景至关重要。 3. **...

    Java Concurrency In Practice.pdf

    书中详细讨论了Java并发工具包中的几个重要并发集合类,如`ConcurrentHashMap`和`CopyOnWriteArrayList`等。 ##### 5.3 阻塞队列与生产者-消费者模式 阻塞队列是一种特殊的队列结构,当队列为空时,从中取元素的...

    《Java Concurrency in Practice》源码

    《Java Concurrency in Practice》是Java并发编程领域的一本经典著作,由Brian Goetz、Tim Peierls、Joshua Bloch、David Holmes和Doug Lea合著。这本书深入探讨了如何在Java环境中有效地设计和实现多线程程序,强调...

    [Java并发编程实践].(Java.Concurrency.in.Practice).Brian.Goetz.英文原版.pdf

    - **并发集合**:Java提供了多种并发集合类,如`ConcurrentHashMap`等,用于解决多线程环境下数据结构的安全访问问题。 - **CompletableFuture**:这是一个强大的异步编程工具,可以简化异步操作的编写过程。书中...

    java-concurrency:Java中的并发

    `java-concurrency`项目可能是对Java并发编程的一个学习资源或者示例集合。 1. **线程与并发基础**:在Java中,线程是并发执行的基本单元。通过创建和启动`Thread`对象,或者使用`Runnable`接口,可以实现多线程。`...

    java concurrency programming

    《Java并发实践》(*Java Concurrency in Practice*)是一本由Brian Goetz、Tim Peierls、Joshua Bloch等专家合著的经典书籍,该书深入浅出地介绍了Java并发编程的基础理论和最佳实践。本书于2006年由Addison Wesley...

Global site tag (gtag.js) - Google Analytics