基于1.7的ConcurrentHashMap

红太狼 2022-01-30 12:27 186阅读 0赞

一个ConcurrentHashMap由一个Segment数组构成，一个Segment由一个HashEntry数组构成，***并且不允许null键和null值***。因此整体的ConcurrentHashMap结构如下：  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RocXR6cQ_size_16_color_FFFFFF_t_70]  
可以说，ConcurrentHashMap是一个二级哈希表，在一个总的哈希表下面，有若干个子哈希表。ConcurrentHashMap的优势就是采用了锁分段技术，每一个Segment就好比一个自治区，读写操作高度自治，Segment之间互不影响。

1.  不同Segment的并发写入：  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RocXR6cQ_size_16_color_FFFFFF_t_70 1]
2.  同一Segment的一写一读：  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RocXR6cQ_size_16_color_FFFFFF_t_70 2]
3.  同一Segment的并发写入：Segment的写入是需要上锁的，因此对同一个Segment的并发写入会被阻塞。  
    ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RocXR6cQ_size_16_color_FFFFFF_t_70 3]  
    由此可见，ConcurrentHashMap当中每个Segment各自持有一把锁，在保证线程安全的同时降低了锁的粒度，让并发操作效率更高。

***ConcurrentHashMap的具体操作如下：***  
Get方法：**注意整个get操作只做了一次hash操作，由于HashEntry中的value属性是用volatile关键字修饰的，保证了内存可见性，所以每次获取时都是最新的值，整个过程不需要加锁。**

1.  对输入的key做Hash运算，得到hash值；
2.  通过hash值，定位到对应的Segment对象；
3.  再次通过hash值，定位到Segment当中数组的具体位置。

***源码如下：***

public V get(Object key) {
            Segment<K,V> s; 
            HashEntry<K,V>[] tab;
            //1，得到hash值
            int h = hash(key);
            //2，通过hash值，定位到对应的Segment对象
            long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
            //3，获取对应的Segment，
            if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null && (tab = s.table) != null) {
            	//4，获取Segment对应的HashEntry<K,V>[]数组，并遍历该数组
                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;
        }

Put方法：

1.  对输入的key做hash运算，得到hash值；
2.  通过hash值，定位到对应的Segment对象；
3.  对该Segment对象进行加锁操作（ReentrantLock可重入锁）；
4.  再次通过hash值，定位到Segment当中数组的具体位置；
5.  插入或覆盖HashEntry对象；
6.  释放锁。  
    ***源码如下：***

public V put(K key, V value) {
            Segment<K,V> s;
            if (value == null)   throw new NullPointerException();
            //1，一次hash运算
            int hash = hash(key);
            //2，根据hash值得到对应的Segment的位移
            int j = (hash >>> segmentShift) & segmentMask;
            //3，确定Segment
            if ((s = (Segment<K,V>)UNSAFE.getObject(segments, (j << SSHIFT) + SBASE)) == null) 
                s = ensureSegment(j);
                //4，对Segment做put操作
            return s.put(key, hash, value, false);
        }
    
    //对Segment做put操作
    final V put(K key, int hash, V value, boolean onlyIfAbsent) {
    			//5，获取可重入锁，只有在调用时它是空闲的才能获取锁；如果锁被其他线程占用，该线程会尝试自旋获取锁，最大次数是64，如果达到最大次数，则改为阻塞获取锁，lock()操作，保证能获取成功。
                HashEntry<K,V> node = tryLock() ? null : scanAndLockForPut(key, hash, value);
                V oldValue;
                try {
                    HashEntry<K,V>[] tab = table;
                    //6，根据hash值定位到Segment当中数组的具体位置
                    int index = (tab.length - 1) & hash;
                    HashEntry<K,V> first = entryAt(tab, index);
                    //7，遍历该链表
                    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 {
                        	//8，该数组位置的元素为null，创建HashEntry对象
                            if (node != null) { node.setNext(first); }
                            else  { node = new HashEntry<K,V>(hash, key, value, first); }
                            int c = count + 1;
                            //9，元素个数达到阀值，考虑扩容，注意扩容扩的是某个Segment对应的HashEntry数组
                            if (c > threshold && tab.length < MAXIMUM_CAPACITY) { 
                            	rehash(node);
                             }
                            //10，不扩容，则put操作
                            else { setEntryAt(tab, index, node); }
                            ++modCount;
                            count = c;
                            oldValue = null;
                            break;
                        }
                    }
                } finally {
                	//11，释放锁
                    unlock();
                }
                //12，返回旧值
                return oldValue;
            }

下面是rehash的代码：参数为新的结点node

private void rehash(HashEntry<K,V> node) {
                HashEntry<K,V>[] oldTable = table;
                int oldCapacity = oldTable.length;
                int newCapacity = oldCapacity << 1;
                threshold = (int)(newCapacity * loadFactor);
                //扩容扩的是某个Segment对应的HashEntry数组
                HashEntry<K,V>[] newTable = (HashEntry<K,V>[]) new HashEntry[newCapacity];
                int sizeMask = newCapacity - 1;
                //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;
                        //2，链表就一个结点，直接放到新数组
                        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);
                            }
                        }
                    }
                }
                //3，添加新节点
                int nodeIndex = node.hash & sizeMask; 
                node.setNext(newTable[nodeIndex]);
                newTable[nodeIndex] = node;
                table = newTable;
            }

***最后一个问题：既然每一个Segment都各自加锁，那么在调用size()方法的时候，怎么解决一致性的问题？***  
size()方法的目的是统计ConcurrentHashMap的总元素，自然需要把各个Segment内部的元素数量汇总起来。但是如果在统计过程中，已经统计过的Segment瞬间插入新的元素，这时候该怎么办？  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RocXR6cQ_size_16_color_FFFFFF_t_70 4]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RocXR6cQ_size_16_color_FFFFFF_t_70 5]  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RocXR6cQ_size_16_color_FFFFFF_t_70 6]  
ConcurrentHashMap的size方式是一个循环嵌套，大致逻辑如下：

1.  遍历所有的Segment;
2.  将Segment的元素数量累加起来；
3.  把Segment的修改次数累加起来；
4.  判断Segment总的修改次数是否大于上次总的修改次数，如果大于，说明统计过程中有修改，重新统计，尝试次数+1；如果相等，说明没有修改，统计结束；
5.  如果尝试次数超过阀值，则对每一个Segment加锁，再重新统计；
6.  再次判断Segment总的修改次数是否等于上次的总的修改次数。由于加锁了，所以一定相等；
7.  释放锁，统计结束。  
    源码如下：

public int size() {
            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 (;;) {
                	//1，重试次数达到RETRIES_BEFORE_LOCK = 2次，对所有Segment加锁，由于retries是从-1开始的，所以加锁之前总共进行了3次尝试
                    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) {
                        	//2，统计总的修改次数和元素个数
                            sum += seg.modCount;
                            int c = seg.count;
                            if (c < 0 || (size += c) < 0)
                                overflow = true;
                        }
                    }
                    //2，last的初始值是0，所以第一次if判断为false,last被赋值为sum,继续循环，如果接下来的循环总的修改次数和上次修改次数相等，则说明该统计过程中没有修改操作，结束统计
                    if (sum == last)
                        break;
                    last = sum;
                }
            } finally {
            	//3，retries在加锁的过程中做了+1操作，所以retries > RETRIES_BEFORE_LOCK；对每个Segment释放锁
                if (retries > RETRIES_BEFORE_LOCK) {
                    for (int j = 0; j < segments.length; ++j)
                        segmentAt(segments, j).unlock();
                }
            }
            //4，返回size
            return overflow ? Integer.MAX_VALUE : size;
        }

为了尽量不锁住所有的Segment，首先会乐观的假设size过程是不会有修改。当尝试2次后，才无奈的转为悲观锁，锁住所有的Segment保证数据的一致性。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RocXR6cQ_size_16_color_FFFFFF_t_70]: /images/20220130/571dc0bfbcdd42368718cc60461ad0ff.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RocXR6cQ_size_16_color_FFFFFF_t_70 1]: /images/20220130/477bd4a0bc2748a0874cffbab7b1c030.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RocXR6cQ_size_16_color_FFFFFF_t_70 2]: /images/20220130/8aeb31597f00431c9c14a40ca0260483.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RocXR6cQ_size_16_color_FFFFFF_t_70 3]: /images/20220130/bc9af41e381141fcbc5342ccb9843269.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RocXR6cQ_size_16_color_FFFFFF_t_70 4]: /images/20220130/981c19005e5643ce91b6b14c642d65dd.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RocXR6cQ_size_16_color_FFFFFF_t_70 5]: /images/20220130/05b77cc8dc324cc08373f968ea6acfaa.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3RocXR6cQ_size_16_color_FFFFFF_t_70 6]: /images/20220130/760c7f566c8c4332897648bc127321bc.png