深入理解ConcurrentHashMap

野性酷女 2022-06-12 04:38 283阅读 0赞

## 说明 ##

HashMap是线程不安全的，在多线程的环境下，操作HashMap会导致线程安全的问题。若使用同步包装器下的HashMap，则会造成很大的性能问题。为此，JDK提供了ConcurrentHashMap来解决此问题。本文通过查看源码和其他优秀博文，对ConcurrentHashMap进行总结。  
由于不同版本中的实现有所不同，本文基于JDK 7进行总结。

## 正文 ##

ConcurrentHashMap采用了分段锁的设计，只有在同一个分段中内才存在竞争关系，所以在进行一般操作时，不需要对整个map加锁。分段锁极大提高了并发下的处理能力。

/*
         * The basic strategy is to subdivide the table among Segments,
         * each of which itself is a concurrently readable hash table.....
         * 由此看出，ConcurrentHashMap底层实现策略是多个段segment，每一个是具有并发可读性的HashTable
      * /

#### ConcurrentHashMap的数据结构 ####

![这里写图片描述][20170716103550461]

#### ConcurrentHashMap的成员变量 ####

static final int DEFAULT_INITIAL_CAPACITY = 16;//初始容量为16
    
    static final float DEFAULT_LOAD_FACTOR = 0.75f;//负载因子为0.75
    
    static final int DEFAULT_CONCURRENCY_LEVEL = 16;
    /*并发度，即同时更新ConcurrentHashMap且不产生锁竞争的最大线程数，也就是分段锁的个数，默认为16.一经指定，便不可改变。如果元素增加导致扩容，也不会增加segment的数量，只会增加segment中数组链表的容量的大小。*/
    
    final Segment<K,V>[] segments;

创建Segement时，采用了延迟初始化机制，每次put操作前都要检查key对应的Segment是否为null，若是则调用ensureSegment()确保对应的Segement被创建。ensureSegment可能在并发环境下被调用，但是并未通过锁机制避免竞争，采用了Unsafe对象的getObjectVolatile（)方法提供的原子读语义结合CAS来确保Segment创建的原子性。

/* all but one segments are constructed only when first needed (see ensureSegment).*/
    
        if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
                    == null) { // recheck
                    Segment<K,V> s = new Segment<K,V>(lf, threshold, tab);
                    while ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
                           == null) {
                        if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s))
                            break;
                    }
                }

Segment是继承自ReentrantLock类，所以Segment对象可以充当锁的角色。每个segment中都由一个HashEntry类型的数组，HashEntry对象构成了数组链表，数组的大小为桶的个数。

static final class Segment<K,V> extends ReentrantLock implements Serializable { 
    
        Segment(float lf, int threshold, HashEntry<K,V>[] tab) {
                this.loadFactor = lf;
                this.threshold = threshold;
                this.table = tab;
            }
          ...
       }

![Segment的内部结构][Segment]

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;
            }
    
            /** * Sets next field with volatile write semantics. (See above * about use of putOrderedObject.) */
            final void setNext(HashEntry<K,V> n) {
                UNSAFE.putOrderedObject(this, nextOffset, n);
            }

在HashEntry类中，hash,key声明为final，value，next声明为volatile。（不同于JDK 6版本中将next设置为final，因此无法在链中添加或删除结点，对于put操作，一律添加的链的头部，对于remove操作，需要将删除结点前的所有结点复制，再将复制完的最后一个结点的next设为当前删除结点的下一个结点。）final的不可变性，volatile的保证可见性是ConcurrentHashMap的get操作不需要同步加锁的重要原因，这里使用setNext方法保证了原子性。

#### 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;
            }
            /* 参数segmentShitft，segmentMask 在定位segment时使用，segmentShift=32-ssize向左移动的次数，segmentMask=ssize-1.ssize的最大长度为65536，对应的segmentShift最大值为16，segmentMask最大值为65536，对应的二进制16位全为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;
        }
    
     public ConcurrentHashMap() {
       //无参构造函数，使用默认值调用三参数的构造函数
            this(DEFAULT_INITIAL_CAPACITY, DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL);
        }

#### Segment中的操作 ####

**put操作**  
在此版本中，put操作使用了自旋锁机制，提高了性能。（因为java线程是映射到系统的线程，阻塞唤醒都需要切换到内核态，频繁的操作会造成严重的性能问题，使用自旋旋可以减轻影响。详见周志明的《深入理解java虚拟机》）

put方法  
进入此方法，尝试获取锁，能获取锁则进行下步操作，不能则调用scanAndLockForput()方法。

获取锁后，如果链中能找到与key相等的节点，并且当前执行的put()方法而不是putIfAbsent()方法，记录旧值，更新该结点的值，退出循环，完成put操作。  
若没找到，此时需创建新的节点，但是在自旋等待时已经创建好，所以不需要创建，只更新它的next指针即可。这里调用了setNext()方法。  
在执行插入操作时，将count值加1，首先会检查本次操作会不会导致segment中节点数量超过阈值threshold，若会，则先进行扩容和rehash操作。（这点与HashMap不同，HashMap的put操作，是先插入，后判断是否超过阈值，决定是否扩容）

final V put(K key, int hash, V value, boolean onlyIfAbsent) {
                HashEntry<K,V> node = tryLock() ? null :
                    scanAndLockForPut(key, hash, value);
                V oldValue;
                try {
                    HashEntry<K,V>[] tab = table;
                    int index = (tab.length - 1) & hash;//定位段中的哪一个桶（segment中HashEntry数组中的某一位置）
                    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 {
                            if (node != null)
                                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方法

当put操作没有获取锁时，不是直接进入等待状态，而是调用此方法，在方法中循环查找链中是否有与key相等的节点，没有，则创建一个新的节点。尝试n次，直到尝试次数超过限制（retries > MAX\_SCAN\_RETRIES 最大尝试次数，单核为1，多核为64），才真正进入等待状态，即所谓的自旋锁等待。

private HashEntry<K,V> scanAndLockForPut(K key, int hash, V value) {
                HashEntry<K,V> first = entryForHash(this, hash);
                HashEntry<K,V> e = first;
                HashEntry<K,V> node = null;
                int retries = -1; // negative while locating node
                while (!tryLock()) {
                    HashEntry<K,V> f; // to recheck first below
                    if (retries < 0) {
                        if (e == null) {
                            if (node == null) // speculatively create node
                                node = new HashEntry<K,V>(hash, key, value, null);
                            retries = 0;
                        }
                        else if (key.equals(e.key))
                            retries = 0;
                        else
                            e = e.next;
                    }
                    else if (++retries > MAX_SCAN_RETRIES) {
                        lock();
                        break;
                    }
                    else if ((retries & 1) == 0 &&
                             (f = entryForHash(this, hash)) != first) {
       //若发现链头发生变化，则更新节点链的链头，重置retries值为-1，重新为尝试获取锁而自旋遍历
                        e = first = f; // re-traverse if entry changed
                        retries = -1;
                    }
                }
                return node;
            }

**rehash操作**  
rehash操作是对某一个段进行操作，创建了一个比原来容量大两倍的数组，然后遍历数组及数组项中的每条链，对于每个节点都重新计算了index，然后创建一个新的节点插入到新的数组中，创建新节点是为了保证其它线程在rehash其间的get操作能够返回正确的值。为了减少新建节点的开销，做了两点优化：1.如果只有一个节点，就直接赋值给新的数组项，若是一个链，先遍历该链找到第一个后面所有节点的index相同的节点p，然后只重新创建节点p以前的节点即可。节点p为头节点的子链直接将p放到新桶中，后面的节点自然就连接上了。

在注释中，我们可以看到，由于扩容是按照2的幂次方进行的，所以新的索引值是原来的或者是原来的加上一个2的幂次方。

@SuppressWarnings("unchecked")
            private void rehash(HashEntry<K,V> node) {
                /* * Reclassify nodes in each list to new table. Because we * are using power-of-two expansion, the elements from * each bin must either stay at same index, or move with a * power of two offset. We eliminate unnecessary node * creation by catching cases where old nodes can be * reused because their next fields won't change. * Statistically, at the default threshold, only about * one-sixth of them need cloning when a table * doubles. The nodes they replace will be garbage * collectable as soon as they are no longer referenced by * any reader thread that may be in the midst of * concurrently traversing table. Entry accesses use plain * array indexing because they are followed by volatile * table write. */
                HashEntry<K,V>[] oldTable = table;
                int oldCapacity = oldTable.length;
                int newCapacity = oldCapacity << 1;
                threshold = (int)(newCapacity * loadFactor);
                HashEntry<K,V>[] newTable =
                    (HashEntry<K,V>[]) new HashEntry[newCapacity];
                int sizeMask = newCapacity - 1;
                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;
            }

**get操作**  
调用get方法，根据key值获得对应的value，进入方法后确定哪个段，哪个桶，再遍历链找出对应的value，不存在返回null。  
与之前版本不同，在JDK 6中，若在链中找到对应的key值，而对应的value值为空，此时需要加锁重新读。此版本用UNSAFE.getObjectVolatile()避免了这个问题。

public V get(Object key) {
            Segment<K,V> s; // manually integrate access methods to reduce overhead
            HashEntry<K,V>[] tab;
            int h = hash(key.hashCode());
            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;
        }

**remove操作**  
调用此方法，跟put方法一样，尝试获取锁，不能则进入自旋锁。  
找到对应value值，直接更改next。在之前版本JDK 6中，需要创建新链以保证并发时不会出现脏读。next设为volatile和调用方法中使用UNSAFE.putOrderedObject()保证了可见性和原子性，避免了脏读的出现。

final V remove(Object key, int hash, Object value) {
                if (!tryLock())
                    scanAndLock(key, hash);
                V oldValue = null;
                try {
                    HashEntry<K,V>[] tab = table;
                    int index = (tab.length - 1) & hash;
                    HashEntry<K,V> e = entryAt(tab, index);
                    HashEntry<K,V> pred = null;
                    while (e != null) {
                        K k;
                        HashEntry<K,V> 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;
            }

**ConcurrentHashMap的操作**  
put()操作  
调用put方法时，先判断value值是否为空。是，抛出异常，由此看出再ConcurrentHashMap中value值不允许为NULL。否，根据key的hashCode值再hash求出hash值，以此hash值再用segmentShift和segmentMask的值定位段（Segment），定位段必须确保段存在，否则调用ensureSegment().  
segment数量是2的n次方，根据hash值的高n位就可以确定在哪一个segment。  
确定segment后，再调用put方法

/* @throws NullPointerException if the specified key or value is null */
    public V put(K key, V value) {
            Segment<K,V> s;
            if (value == null)
                throw new NullPointerException();
            int hash = hash(key.hashCode());
            int j = (hash >>> segmentShift) & segmentMask;
            if ((s = (Segment<K,V>)UNSAFE.getObject          // nonvolatile; recheck
                 (segments, (j << SSHIFT) + SBASE)) == null) // in ensureSegment
                s = ensureSegment(j);
            return s.put(key, hash, value, false);
        }
       //由此方法可以看出ConcurrentHashMap中key或value值都不允许为null

ConcurrentHashMap中的isEmpty() ,size(),containsValue(),contains()操作都需要全局扫描Map，为减少锁对性能的影响，这里循环查找，并计算两次modCount值，若两次相等，说明两次遍历的过程中，整个的Map没有发生改变，查找的值是正确的，否则将segment逐个加锁计算。

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<K,V>[] 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<K,V> 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;
        }

putIfAbsent、replace、remove、clear操作与put方法类似，它们在Segment中都实现，只需要通过hash值找到Segment，然后调用相应方法即可。

[20170716103550461]: /images/20220612/791a2a3c297b4e39ad3c6ea88eebe831.png
[Segment]: /images/20220612/10706ecca0fa4b31a9f94530bd987fb2.png