并发容器之ConcurrentHashMap源码实现 我会带着你远行 2022-04-25 10:34 179阅读 0赞 ## 并发容器之ConcurrentHashMap源码实现 ## ### 一、背景 ### 其实这一小节主要内容可以换一种说法：“为什么要使用ConcurrentHashMap,而非HashMap?尤其是在多线程环境下？” 其实我简单总结了以下三点原因，可以回答上述问题。此外声明下本文研究的源码版本以JDK1.7标准，其他版本的代码略有差异，尤其是JDK1.8 ,后面有时间在详细研究。 1. **HashMap的线程不安全** 在多线程环境下，使用HashMap进行put操作会引起死循环，导致CPU利用率接近100%，所 以在并发情况下不能使用HashMap。 HashMap在并发执行put操作时会引起死循环，是因为多线程会导致HashMap的Entry链表 形成环形数据结构，一旦形成环形数据结构，Entry的next节点永远不为空，就会产生死循环获 取Entry。 2. **效率低下的HashTable** HashTable容器使用synchronized来保证线程安全，但在线程竞争激烈的情况下HashTable 的效率非常低下。因为当一个线程访问HashTable的同步方法，其他线程也访问HashTable的同 步方法时，会进入阻塞或轮询状态。如线程1使用put进行元素添加，线程2不但不能使用put方 法添加元素，也不能使用get方法来获取元素，所以竞争越激烈效率越低。 3. **ConcurrentHashMap的锁分段技术可有效提升并发访问率** HashTable容器在竞争激烈的并发环境下表现出效率低下的原因是所有访问HashTable的 线程都必须竞争同一把锁，假如容器里有多把锁，每一把锁用于锁容器其中一部分数据，那么 当多线程访问容器里不同数据段的数据时，线程间就不会存在锁竞争，从而可以有效提高并 发访问效率，这就是ConcurrentHashMap所使用的锁分段技术。首先将数据分成一段一段地存 储，然后给每一段数据配一把锁，当一个线程占用锁访问其中一个段数据的时候，其他段的数 据也能被其他线程访问。 ### 二、ConcurrentHashMap实现原理 ### #### 2.1 原理概述 #### 正如第一节中描述的3个理由，其实其中已经包括了其原理描述。 HashTable容器在竞争激烈的并发环境下表现出效率低下的原因是所有访问HashTable的 线程都必须竞争同一把锁，假如容器里有多把锁，每一把锁用于锁容器其中一部分数据，那么 当多线程访问容器里不同数据段的数据时，线程间就不会存在锁竞争，从而可以有效提高并 发访问效率，这就是ConcurrentHashMap所使用的锁分段技术。首先将数据分成一段一段地存 储，然后给每一段数据配一把锁，当一个线程占用锁访问其中一个段数据的时候，其他段的数 据也能被其他线程访问。 暂时先这样理解，下面在解析ConcurrentHashMap的属性、构造器及内部类的时候，你就会明白上面的那段话。 #### 2.2 原理图 #### ![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTcyMzU0NA_size_16_color_FFFFFF_t_70] 记住这张图的样子，在后面的操作的逻辑可以大概联想这张图进行思考，应该会顺畅很多 ### 三、源码分析 ### #### 3.1 属性 #### /* ---------------- Constants:一些系统默认的常量 -------------- */ // 默认初始容量大小 static final int DEFAULT_INITIAL_CAPACITY = 16; // 默认负载因子 static final float DEFAULT_LOAD_FACTOR = 0.75f; // 默认的segment容量大小 static final int DEFAULT_CONCURRENCY_LEVEL = 16; // 最大容量 static final int MAXIMUM_CAPACITY = 1 << 30; // 最小segment容量大小 static final int MIN_SEGMENT_TABLE_CAPACITY = 2; // 最小segment容量大小。也就是65535 static final int MAX_SEGMENTS = 1 << 16; // slightly conservative // 在锁定之前重试的次数（内部类HashEntry中使用） static final int RETRIES_BEFORE_LOCK = 2; /* ---------------- Fields:属性 -------------- */ private transient final int hashSeed = randomHashSeed(this); // segmentMask是散列运算的掩码，等于ssize减1 // 掩码的二进制各个位的值都是1。 final int segmentMask; // segmentShift用于定位参与散列运算的位数 final int segmentShift; // segments数组 final Segment<K,V>[] segments; // key集合 transient Set<K> keySet; // 实体集合 transient Set<Map.Entry<K,V>> entrySet; // value集合 transient Collection<V> values; #### 3.2 构造器 #### // 思路：通过initialCapacity、loadFactor和concurrencyLevel等几个 // 参数来初始化segment数组、段偏移量segmentShift、段掩码segmentMask和每个segment里的 // HashEntry数组。 // 步骤： // 1. 初始化一些属性值 // 1.1 计算segments数组的长度ssize的值。通过concurrencyLevel计算得出的。为了能 // 通过按位与的散列算法来定位segments数组的索引，必须保证segments数组的长度是2的N次方 // （power-of-two size），所以必须计算出一个大于或等于concurrencyLevel的最小的2的N次方值 // 来作为segments数组的长度。假如concurrencyLevel等于14、15或16，ssize都会等于16，即容器里锁的个数也是16。 // concurrencyLevel的最大值是65535，这意味着segments数组的长度最大为65536，对应的二进制是16位。 // 1.2 初始化segmentShift和segmentMask。 // 这两个全局变量需要在定位segment时的散列算法里使用，sshift等于ssize从1向左移位的 // 次数，在默认情况下concurrencyLevel等于16，1需要向左移位移动4次，所以sshift等于4。 // segmentShift用于定位参与散列运算的位数，segmentShift等于32减sshift，所以等于28，这里之所 // 以用32是因为ConcurrentHashMap里的hash()方法输出的最大数是32位的，后面的测试中我们 // 可以看到这点。segmentMask是散列运算的掩码，等于ssize减1，即15，掩码的二进制各个位的 // 值都是1。因为ssize的最大长度是65536，所以segmentShift最大值是16，segmentMask最大值是 // 65535，对应的二进制是16位，每个位都是1。 // 2. 初始化Segment[]对象 // 2.1 初始化Segment对象 // 输入参数initialCapacity是ConcurrentHashMap的初始化容量，loadfactor是每个segment的负 // 载因子，在构造方法里需要通过这两个参数来初始化数组中的每个segment。 // 变量cap就是segment里HashEntry数组的长度，它等于initialCapacity除以ssize // 的倍数c，如果c大于1，就会取大于等于c的2的N次方值，所以cap不是1，就是2的N次方。 // segment的容量threshold＝（int）cap*loadFactor，默认情况下initialCapacity等于16，loadfactor等于 // 0.75，通过运算cap等于1，threshold等于零。 // 2.2 初始化Segment[]对象 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] // 创建segments数组并初始化第一个Segment，其余的Segment延迟初始化 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(int initialCapacity, float loadFactor) { this(initialCapacity, loadFactor, DEFAULT_CONCURRENCY_LEVEL); } public ConcurrentHashMap(int initialCapacity) { this(initialCapacity, DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL); } public ConcurrentHashMap() { this(DEFAULT_INITIAL_CAPACITY, DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL); } public ConcurrentHashMap(Map<? extends K, ? extends V> m) { this(Math.max((int) (m.size() / DEFAULT_LOAD_FACTOR) + 1, DEFAULT_INITIAL_CAPACITY), DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL); putAll(m); } #### 3.3 定位Segment #### 既然ConcurrentHashMap使用分段锁Segment来保护不同段的数据，那么在插入和获取元素 的时候，必须先通过散列算法定位到Segment。可以看到ConcurrentHashMap会首先使用 Wang/Jenkins hash的变种算法对元素的hashCode进行一次再散列。 private int hash(Object k) { int h = hashSeed; if ((0 != h) && (k instanceof String)) { return sun.misc.Hashing.stringHash32((String) k); } h ^= k.hashCode(); // Spread bits to regularize both segment and index locations, // using variant of single-word Wang/Jenkins hash. h += (h << 15) ^ 0xffffcd7d; h ^= (h >>> 10); h += (h << 3); h ^= (h >>> 6); h += (h << 2) + (h << 14); return h ^ (h >>> 16); } 之所以进行再散列，目的是减少散列冲突，使元素能够均匀地分布在不同的Segment上， 从而提高容器的存取效率。假如散列的质量差到极点，那么所有的元素都在一个Segment中， 不仅存取元素缓慢，分段锁也会失去意义。笔者做了一个测试，不通过再散列而直接执行散列 计算。 System.out.println(Integer.parseInt("0001111", 2) & 15); System.out.println(Integer.parseInt("0011111", 2) & 15); System.out.println(Integer.parseInt("0111111", 2) & 15); System.out.println(Integer.parseInt("1111111", 2) & 15); 计算后输出的散列值全是15，通过这个例子可以发现，如果不进行再散列，散列冲突会非 常严重，因为只要低位一样，无论高位是什么数，其散列值总是一样。我们再把上面的二进制 数据进行再散列后结果如下（为了方便阅读，不足32位的高位补了0，每隔4位用竖线分割下）。 0100｜0111｜0110｜0111｜1101｜1010｜0100｜1110 1111｜0111｜0100｜0011｜0000｜0001｜1011｜1000 0111｜0111｜0110｜1001｜0100｜0110｜0011｜1110 1000｜0011｜0000｜0000｜1100｜1000｜0001｜1010 可以发现，每一位的数据都散列开了，通过这种再散列能让数字的每一位都参加到散列 运算当中，从而减少散列冲突。ConcurrentHashMap通过以下散列算法定位segment。 private Segment<K,V> segmentForHash(int h) { long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE; return (Segment<K,V>) UNSAFE.getObjectVolatile(segments, u); } 默认情况下segmentShift为28，segmentMask为15，再散列后的数最大是32位二进制数据， 向右无符号移动28位，意思是让高4位参与到散列运算中，（hash>>>segmentShift） &segmentMask的运算结果分别是4、15、7和8，可以看到散列值没有发生冲突。 #### 3.4 核心内部类-Segment #### Segment类是一类特殊的hash表，继承了ReentrantLock类具备了锁的功能。这是ConcurrentHashMap中最核心的内部类，put，get, remove的核心逻辑都在其中。 static final class Segment<K,V> extends ReentrantLock implements Serializable { private static final long serialVersionUID = 2249069246763182397L; /** * 总论： * 1.segment的读操作不需要加锁，但需要volatile读 * 2.当进行扩容时(调用reHash方法)，需要拷贝原始数据，在拷贝数据上操作，保证在扩容完成前读操作仍可以在原始数据上进行。 * 3.只有引起数据变化的操作需要加锁。 * 4.scanAndLock(删除、替换)/scanAndLockForPut(新增)两个方法提供了获取锁的途径，是通过自旋锁实现的。 * 5.在等待获取锁的过程中，两个方法都会对目标数据进行查找，每次查找都会与上次查找的结果对比，虽然查找结果不会被调用它的方法使用，但是这样做可以减少后续操作可能的cache miss。 * */ // 自旋锁的等待次数上限，多处理器时64次，单处理器时1次。 // 每次等待都会进行查询操作，当等待次数超过上限时，不再自旋，调用lock方法等待获取锁 static final int MAX_SCAN_RETRIES = Runtime.getRuntime().availableProcessors() > 1 ? 64 : 1; // Segment中的hash表，与hashMap结构相同，表中每个元素都是一个链表 transient volatile HashEntry<K,V>[] table; // 表中元素个数 transient int count; // 记录数据变化操作的次数. // 作用：这一数值主要为Map的isEmpty和size方法提供同步操作检查，这两个方法没有为全表加锁。 // 在统计segment.count前后，都会统计segment.modCount，如果前后两次值发生变化，可以判断在统计count期间有segment发生了其它操作 transient int modCount; // 容量阈值，超过这一数值后segment将进行扩容，容量变为原来的两倍 // threshold = loadFactor * table.length transient int threshold; final float loadFactor; Segment(float lf, int threshold, HashEntry<K,V>[] tab) { this.loadFactor = lf; this.threshold = threshold; this.table = tab; } // onlyIfAbsent:若为true，当key已经有对应的value时，不进行替换；若为false，即使key已经有对应的value，仍进行替换。 // 关于put方法，很重要的一点是segment最大长度的问题： // 代码 c > threshold && tab.length < MAXIMUM_CAPACITY 作为是否需要扩容的判断条件。 // 扩容条件是node总数超过阈值且table长度小于MAXIMUM_CAPACITY也就是2的30次幂。 // 由于扩容都是容量翻倍，所以tab.length最大值就是2的30次幂。此后，即使node总数超过了阈值，也不会扩容了。 // 由于table[n]对应的是一个链表，链表内元素个数理论上是无限的，所以segment的node总数理论上也是无上限的。 // ConcurrentHashMap的size()方法考虑到了这个问题，当计算结果超过Integer.MAX_VALUE时，直接返回Integer.MAX_VALUE. final V put(K key, int hash, V value, boolean onlyIfAbsent) { // tryLock判断是否已经获得锁. // 如果没有获得，调用scanAndLockForPut方法自旋等待获得锁。 HashEntry<K,V> node = tryLock() ? null : scanAndLockForPut(key, hash, value); V oldValue; try { HashEntry<K,V>[] tab = table; // 计算key在表中的下标 int index = (tab.length - 1) & hash; // 获取链表的第一个node HashEntry<K,V> first = entryAt(tab, index); for (HashEntry<K,V> e = first;;) { // 链表下一个node不为空，比较key值是否相同 // 相同的，根据onlyIfAbsent决定是否替换已有的值 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 { // 链表遍历到最后一个node，仍没有找到key值相同的. // 此时应当生成新的node，将node的next指向链表表头，这样新的node将处于链表的【表头】位置 if (node != null) node.setNext(first); else // node为null，表明sca 表明scanAndLockForPut过程中找到了key值相同的nodenAndLockForPut过程中找到了key值相同的node // 可以断定在等待获取锁的过程中，这个node被删除了，此时需要新建一个node 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; } @SuppressWarnings("unchecked") private void rehash(HashEntry<K,V> node) { // 拷贝table，所有操作都在oldTable上进行，不会影响无需获得锁的读操作 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; // 新的table下标，定位链表 int idx = e.hash & sizeMask; if (next == null) // Single node on list // 链表只有一个node，直接赋值 newTable[idx] = e; else { // Reuse consecutive sequence at same slot // 这里获取特殊node 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; } } // 步骤一中的table[n]赋值过程 newTable[lastIdx] = lastRun; // Clone remaining nodes // 步骤二，遍历剩余node，插入对应表头 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; } // put方法调用本方法获取锁，通过自旋锁等待其他线程释放锁。 // 变量retries记录自旋锁循环次数，当retries超过MAX_SCAN_RETRIES时，不再自旋，调用lock方法等待锁释放。 // 变量first记录hash计算出的所在链表的表头node，每次循环结束，重新获取表头node，与first比较，如果发生变化，说明在自旋期间，有新的node插入了链表，retries计数重置。 // 自旋过程中，会遍历链表，如果发现不存在对应key值的node，创建一个，这个新node可以作为返回值返回。 // 根据官方注释，自旋过程中遍历链表是为了缓存预热，减少hash表经常出现的cache miss 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仍没有找到匹配 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; } // 比较first与新获得的链表表头node是否一致，如果不一致，说明该链表别修改过，自旋计数重置 else if ((retries & 1) == 0 && (f = entryForHash(this, hash)) != first) { e = first = f; // re-traverse if entry changed retries = -1; } } return node; } // remove,replace方法会调用本方法获取锁，通过自旋锁等待其他线程释放锁。 // 与scanAndLockForPut机制相似。 private void scanAndLock(Object key, int hash) { // similar to but simpler than scanAndLockForPut HashEntry<K,V> first = entryForHash(this, hash); HashEntry<K,V> e = first; int retries = -1; while (!tryLock()) { HashEntry<K,V> f; if (retries < 0) { if (e == null || 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) { e = first = f; retries = -1; } } } // 删除key-value都匹配的node，删除过程很简单 // 1.根据hash计算table下标index // 2.根据index定位链表，遍历链表node，如果存在node的key值和value值都匹配，删除该node。 // 3.令node的前一个节点pred的pred.next = node.next。 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; } // 和remove()过程相似 final boolean replace(K key, int hash, V oldValue, V newValue) { if (!tryLock()) scanAndLock(key, hash); boolean replaced = false; try { HashEntry<K,V> e; for (e = entryForHash(this, hash); e != null; e = e.next) { K k; if ((k = e.key) == key || (e.hash == hash && key.equals(k))) { if (oldValue.equals(e.value)) { e.value = newValue; ++modCount; replaced = true; } break; } } } finally { unlock(); } return replaced; } // 和remove()过程相似 final V replace(K key, int hash, V value) { if (!tryLock()) scanAndLock(key, hash); V oldValue = null; try { HashEntry<K,V> e; for (e = entryForHash(this, hash); e != null; e = e.next) { K k; if ((k = e.key) == key || (e.hash == hash && key.equals(k))) { oldValue = e.value; e.value = value; ++modCount; break; } } } finally { unlock(); } return oldValue; } final void clear() { lock(); try { HashEntry<K,V>[] tab = table; for (int i = 0; i < tab.length ; i++) setEntryAt(tab, i, null); ++modCount; count = 0; } finally { unlock(); } } } #### 3.5 常见操作 #### 1. get() Segment的get操作实现非常简单和高效。先经过一次再散列，然后使用这个散列值通过散 列运算定位到Segment，再通过散列算法定位到元素，代码如下。 public V get(Object key) { Segment<K,V> s; // manually integrate access methods to reduce overhead HashEntry<K,V>[] tab; int h = hash(key); // 定位到key所属的segment在内存中偏移量 long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE; if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null && (tab = s.table) != null) { // 定位到key在segment对象中某个hash位置，循环此链表进行判断：发现key值相等直接返回。 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; } get()是高效的。原因有以下几点来保证： 1. volatile保证了变量的可见性 2. get()是无锁的，没有锁资源消耗。 3. 强大的hash散列的支持，使其在性能趋近O(1)的时空复杂度 2. put() 由于put方法里需要对共享变量进行写入操作，所以为了线程安全，在操作共享变量时必 须加锁。put方法首先定位到Segment，然后在Segment里进行插入操作。插入操作需要经历两个 步骤，第一步判断是否需要对Segment里的HashEntry数组进行扩容，第二步定位添加元素的位 置，然后将其放在HashEntry数组里。代码如下： public V put(K key, V value) { // 1. 定位Segment Segment<K,V> s; if (value == null) throw new NullPointerException(); int hash = hash(key); int j = (hash >>> segmentShift) & segmentMask; if ((s = (Segment<K,V>)UNSAFE.getObject // nonvolatile; recheck (segments, (j << SSHIFT) + SBASE)) == null) // in ensureSegment s = ensureSegment(j); // 2. 调用的segment的put()方法进行放置 // 具体步骤可参见上面Segment部门 return s.put(key, hash, value, false); } Segment中put的主要思路分为两部分： 1. 是否需要扩容 在插入元素前会先判断Segment里的HashEntry数组是否超过容量（threshold），如果超过阈 值，则对数组进行扩容。值得一提的是，Segment的扩容判断比HashMap更恰当，因为HashMap 是在插入元素后判断元素是否已经到达容量的，如果到达了就进行扩容，但是很有可能扩容 之后没有新元素插入，这时HashMap就进行了一次无效的扩容。 1. 如何扩容 在扩容的时候，首先会创建一个容量是原来容量两倍的数组，然后将原数组里的元素进 行再散列后插入到新的数组里。为了高效，ConcurrentHashMap不会对整个容器进行扩容，而只 对某个segment进行扩容。 3. remove() remove()方法的实现的比较简单，和之前get和put方法类似，先定位Segment的位置。然后调用Segmenet中的删除操作。代码如下： public V remove(Object key) { int hash = hash(key); Segment<K,V> s = segmentForHash(hash); return s == null ? null : s.remove(key, hash, null); } 4. size() 如果要统计整个ConcurrentHashMap里元素的大小，就必须统计所有Segment里元素的大小 后求和。Segment里的全局变量count是一个volatile变量，那么在多线程场景下，是不是直接把 所有Segment的count相加就可以得到整个ConcurrentHashMap大小了呢？不是的，虽然相加时 可以获取每个Segment的count的最新值，但是可能累加前使用的count发生了变化，那么统计结 果就不准了。所以，最安全的做法是在统计size的时候把所有Segment的put、remove和clean方法 全部锁住，但是这种做法显然非常低效。 因为在累加count操作过程中，之前累加过的count发生变化的几率非常小，所以 ConcurrentHashMap的做法是先尝试2次通过不锁住Segment的方式来统计各个Segment大小，如 果统计的过程中，容器的count发生了变化，则再采用加锁的方式来统计所有Segment的大小。 那么ConcurrentHashMap是如何判断在统计的时候容器是否发生了变化呢？使用modCount 变量，在put、remove和clean方法里操作元素前都会将变量modCount进行加1，那么在统计size 前后比较modCount是否发生变化，从而得知容器的大小是否发生变化。 [watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zOTcyMzU0NA_size_16_color_FFFFFF_t_70]: /images/20220125/f4624fc068ea45bfb7d56a3b0c62d7cd.png
还没有评论，来说两句吧...