JDK1.7 ConcurrentHashMap实现原理浅析

在多线程场景下使用HashMap会造成死循环，CPU100%等问题，所以我们不能在多线程场景下使用HashMap，另外一个集合类HashTable是线程安全的，但其使用synchronized这种粗粒度的锁来实现的，所以并发场景下性能低下，在多线程（并发）场景下我们推荐使用ConcurrentHashMap类。 这里放一张ConcurrentHashMap的类图：

可以看出该类也实现了Map接口，所以通常可以直接替换HashMap使用而不用修改业务代码。

HashTable之所以性能低下，原因是多线程竞争同一把锁（HashTable粗暴的为整个存储结构加了锁），而ConcurrentHashMap则改进这了一点。该类通过分段加锁来降低资源竞争，底层的存储数组结构不再像HashMap一样直接是一个哈希表（数组），而是使用Segment数组来实现分片，Segment类继承了ReentrantLock类，所以它本身也是一个可重入锁，每个Segment则相当于一个HashMap，同样使用哈希表存储数据，每个Bucket都是一个链表，其内部实现思想与HashMap基本一致，不同的是put、remove等方法都是加了锁的。这样分段加锁的好处是，如果两个线程操作的不是同一个Segment，则相互不影响，不用相互等待，从而提升了性能。

Segment数组本身是不加锁的，那么在向ConcurrentHashMap中添加元素时，会根据键计算出的HashCode来定位Segment，这个过程因为不涉及修改操作，所以不需要加锁。而针对特定的Segment内部数据进行操作，则需要加锁，下面以JDK1.7版ConcurrentHashMap源码为例进行解读。

JDK1.7 ConcurrentHashMap源码解读

ConcurrentHashMap底层实现涉及多个内部类，这里简述一下

• HashEntry类

static final class HashEntry
    
      {        final int hash;        final K key;        volatile V value;        volatile HashEntry
     
       next;        // ... ...    }
     
    

• Segment类（这里删除了代码细节和注释）

static final class Segment
    
      extends ReentrantLock implements Serializable {        static final int MAX_SCAN_RETRIES =            Runtime.getRuntime().availableProcessors() > 1 ? 64 : 1;        transient volatile HashEntry
     
      [] table;        transient int count;        transient int modCount;        transient int threshold;        final float loadFactor;        Segment(float lf, int threshold, HashEntry
      
       [] tab) {}        final V put(K key, int hash, V value, boolean onlyIfAbsent) {}        @SuppressWarnings("unchecked")        private void rehash(HashEntry
       
         node) {}        private HashEntry
        
          scanAndLockForPut(K key, int hash, V value) {}        private void scanAndLock(Object key, int hash) {}        final V remove(Object key, int hash, Object value) {}        final boolean replace(K key, int hash, V oldValue, V newValue) {}        final V replace(K key, int hash, V value) {}        final void clear() {}    }
        
       
      
     
    

ConcurrentHashMap中分段是由Segment数组实现的，而每个Segment的内部存储结构为哈希表（数组），而每个Bucket则是由HashEntry构成的链表组成（这点与HashMap是一样的）。

下面通过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;        // 找到刚好比 concurrencyLevel 大或相等的2的整数次幂        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;        // 计算每段容量（取刚好大于等于c的2的整数次幂）        int cap = MIN_SEGMENT_TABLE_CAPACITY;        while (cap < c)            cap <<= 1;        // create segments and segments[0]        Segment
    
      s0 =            new Segment
     
      (loadFactor, (int)(cap * loadFactor),                             (HashEntry
      
       [])new HashEntry[cap]);        Segment
       
        [] ss = (Segment
        
         [])new Segment[ssize];        UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]        this.segments = ss;    }
        
       
      
     
    

与HashMap不同该类的构造方法多了一个concurrencyLevel参数，该参数主要用于控制分段数，该类的其它构造方法都脱胎与该方法，这里不再赘述，其中无参构造方法中的参数默认值分别是：initialCapacity=16、loadFactor=0.75f、concurrencyLevel=16。 构造方法中分别初始化了：分段数、每段容器大小、Segment数组和第一个Segment节点。

isEmpty和size方法

public boolean isEmpty() {        long sum = 0L;        final Segment
    
     [] segments = this.segments;        for (int j = 0; j < segments.length; ++j) {            Segment
     
       seg = segmentAt(segments, j);            if (seg != null) {                if (seg.count != 0)                    return false;                sum += seg.modCount;            }        }        if (sum != 0L) { // recheck unless no modifications            for (int j = 0; j < segments.length; ++j) {                Segment
      
        seg = segmentAt(segments, j);                if (seg != null) {                    if (seg.count != 0)                        return false;                    sum -= seg.modCount;                }            }            if (sum != 0L)                return false;        }        return true;    }    public int size() {        // Try a few times to get accurate count. On failure due to        // continuous async changes in table, resort to locking.        final Segment
       
        [] segments = this.segments;        int size;        boolean overflow; // true if size overflows 32 bits        long sum;         // sum of modCounts        long last = 0L;   // previous sum        int retries = -1; // first iteration isn't retry        try {            for (;;) {                if (retries++ == RETRIES_BEFORE_LOCK) {                    for (int j = 0; j < segments.length; ++j)                        ensureSegment(j).lock(); // force creation                }                sum = 0L;                size = 0;                overflow = false;                for (int j = 0; j < segments.length; ++j) {                    Segment
        
          seg = segmentAt(segments, j);                    if (seg != null) {                        sum += seg.modCount;                        int c = seg.count;                        if (c < 0 || (size += c) < 0)                            overflow = true;                    }                }                if (sum == last)                    break;                last = sum;            }        } finally {            if (retries > RETRIES_BEFORE_LOCK) {                for (int j = 0; j < segments.length; ++j)                    segmentAt(segments, j).unlock();            }        }        return overflow ? Integer.MAX_VALUE : size;    }
        
       
      
     
    

两个实现方法的思路相同，都是遍历全部Segment，再计算每个Segment内部元素个数。需要注意的是为了防止在方法执行过程中，Segment本身会发生变化（如：添加、删除元素等），但遍历过程中对Segment加锁，方法执行结束后释放锁，所以这两个方法的性能不如HashMap的高（应用场景不同，本身也没什么可比性）。

put、putIfAbsent方法

public V put(K key, V value) {        Segment
    
      s;        if (value == null)            throw new NullPointerException();        int hash = hash(key);        int j = (hash >>> segmentShift) & segmentMask;        if ((s = (Segment
     
      )UNSAFE.getObject          // nonvolatile; recheck             (segments, (j << SSHIFT) + SBASE)) == null) //  in ensureSegment            s = ensureSegment(j);        return s.put(key, hash, value, false);    }    public V putIfAbsent(K key, V value) {        Segment
      
        s;        if (value == null)            throw new NullPointerException();        int hash = hash(key);        int j = (hash >>> segmentShift) & segmentMask;        if ((s = (Segment
       
        )UNSAFE.getObject             (segments, (j << SSHIFT) + SBASE)) == null)            s = ensureSegment(j);        return s.put(key, hash, value, true);    }    static final class Segment
        
          extends ReentrantLock implements Serializable {        final V put(K key, int hash, V value, boolean onlyIfAbsent) {            HashEntry
         
           node = tryLock() ? null : scanAndLockForPut(key, hash, value); V oldValue; try { HashEntry
          
           [] tab = table; int index = (tab.length - 1) & hash; HashEntry
           
             first = entryAt(tab, index); for (HashEntry
            
              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
             
              (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; } }
             
            
           
          
         
        
       
      
     
    

put方法的逻辑比较深，但有HashMap的源码基础的话，其实也不复杂。在ConcurrentHashMap中的put方法实际上只是根据HashCode找到对应的Segment，这个过程不需要加锁，而实际put动作是由Segment类中的put方法完成的。 该方法相比HashMap中的put方法，只是增加了锁的机制（毕竟是面向多线程场景）。

containsKey、containsValue、contains方法

public boolean containsKey(Object key) {        Segment
    
      s; // same as get() except no need for volatile value read        HashEntry
     
      [] tab;        int h = hash(key);        long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;        if ((s = (Segment
      
       )UNSAFE.getObjectVolatile(segments, u)) != null &&            (tab = s.table) != null) {            for (HashEntry
       
         e = (HashEntry
        
         ) 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 true;            }        }        return false;    }
        
       
      
     
    

只是简单的查找，与size不同的是，不需要加锁（确实也没有加锁的必要，如果元素存在则不再添加，可以使用putIfAbsent方法）。

get方法

public V get(Object key) {        Segment
    
      s; // manually integrate access methods to reduce overhead        HashEntry
     
      [] tab;        int h = hash(key);        long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;        if ((s = (Segment
      
       )UNSAFE.getObjectVolatile(segments, u)) != null &&            (tab = s.table) != null) {            for (HashEntry
       
         e = (HashEntry
        
         ) 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;    }
        
       
      
     
    

remove方法

public V remove(Object key) {        int hash = hash(key);        Segment
    
      s = segmentForHash(hash);        return s == null ? null : s.remove(key, hash, null);    }    static final class Segment
     
       extends ReentrantLock implements Serializable {        final V remove(Object key, int hash, Object value) {            if (!tryLock())                scanAndLock(key, hash);            V oldValue = null;            try {                HashEntry
      
       [] tab = table;                int index = (tab.length - 1) & hash;                HashEntry
       
         e = entryAt(tab, index);                HashEntry
        
          pred = null;                while (e != null) {                    K k;                    HashEntry
         
           next = e.next; if ((k = e.key) == key || (e.hash == hash && key.equals(k))) { V v = e.value; if (value == null || value == v || value.equals(v)) { if (pred == null) setEntryAt(tab, index, next); else pred.setNext(next); ++modCount; --count; oldValue = v; } break; } pred = e; e = next; } } finally { unlock(); } return oldValue; } }
         
        
       
      
     
    

结语

偷懒了，偷懒了，最近天天看源码，看得头大，这篇就到这里了（草草结束），主要是理解实现原理，后面再完善细节吧。