ConcurrentHashMap源码解析-JDK1.7-6ren

ConcurrentHashMap源码解析-JDK1.7

转载作者：知者更新时间：2024-03-13 06:32:03

ConcurrentHashMap概述

支持检索的完全并发和可调整的预期更新并发的哈希表。
此类遵循与 Hashtable 相同的功能规范，并包含与 Hashtable 的每个方法对应的方法版本。
但是，即使所有操作都是线程安全的，检索操作也不需要锁定，并且不支持以阻止所有访问的方式锁定整个表。
在依赖线程安全但不依赖同步细节的程序中，此类与 Hashtable 完全可互操作。检索操作（包括 get）一般不会阻塞，因此可能与更新操作（包括 put 和 remove）重叠。
检索反映了最近完成的更新操作在其开始时保持的结果。对于 putAll 和 clear 等聚合操作，并发检索可能仅反映插入或删除某些条目。类似地，Iterators 和 Enumerations 返回的元素反映了在 iteratorenumeration 创建之时或之后的某个时刻哈希表的状态。
它们不会抛出 ConcurrentModificationException。
但是，迭代器被设计为一次只能由一个线程使用。更新操作之间允许的并发性由可选的 concurrencyLevel 构造函数参数（默认为 16）指导，该参数用作内部大小调整的提示。
该表在内部进行了分区，以尝试允许指定数量的并发更新而不会发生争用。
因为哈希表中的放置本质上是随机的，所以实际的并发性会有所不同。理想情况下，您应该选择一个值来容纳尽可能多的线程同时修改表。使用显着高于您需要的值会浪费空间和时间，而显着降低的值会导致线程争用。
但是一个数量级内的高估和低估通常不会产生太大的影响。当已知只有一个线程会修改而所有其他线程只会读取时，值 1 是合适的。此外，调整这种或任何其他类型的哈希表的大小是一个相对较慢的操作，因此，如果可能，最好在构造函数中提供预期表大小的估计值。
不允许null键和null值。

JDK 1.7

基本策略是将表细分为 Segments，每个 Segment本身就是一个并发可读的哈希表。为了减少占用空间，仅在第一次需要时才构建除一个段之外的所有段（请参阅 ensureSegment）。(segments[0]是在初始化ConcurrentHashMap时就构建好的)
为了在存在惰性构造的情况下保持可见性，对段以及段表元素的访问必须使用 volatile 访问，这是通过下面的方法 segmentAt 等中的 Unsafe 完成的。这些提供了 AtomicReferenceArrays 的功能，但降低了间接级别。此外，锁定操作中的表元素和条目“next”字段的volatile-writes使用更便宜的“lazySet”写入形式（通过 putOrderedObject），因为这些写入始终伴随着锁释放，以保持表更新的顺序一致性。

默认参数

// 默认容量
static final int DEFAULT_INITIAL_CAPACITY = 16;
// 默认加载因子
static final float DEFAULT_LOAD_FACTOR = 0.75f;
// 默认的并发级别
 static final int DEFAULT_CONCURRENCY_LEVEL = 16;
// 最大容量
static final int MAXIMUM_CAPACITY = 1 << 30;
// 每段表的最小容量。必须是 2 的幂，至少是 2 ，以避免在懒惰构造后立即调整大小
static final int MIN_SEGMENT_TABLE_CAPACITY = 2;
// 允许的最大段数；用于绑定构造函数参数。必须是小于 1 << 24 的 2 的幂。
static final int MAX_SEGMENTS = 1 << 16; 

// 段进行索引的掩码值，key的hashCode的高位用于选择段
final int segmentMask;
// 段内索引的移位值
final int segmentShift;
// 分段数组，每一个段都是一个hash表
final Segment<K,V>[] segments;

构造方法

初始化操作：段数组、段数组第一个元素，段偏移量

// 指定并发级别的构造方法
    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; // 段内数组的长度，得到2的整数幂的大小
        while (ssize < concurrencyLevel) {
            ++sshift;
            ssize <<= 1;
        }
        this.segmentShift = 32 - sshift; // 段偏移量,默认初始化 segmentShift=32-4=28
        this.segmentMask = ssize - 1; // 段掩码值，默认初始化 segmentMask=16-1=15
        if (initialCapacity > MAXIMUM_CAPACITY)
            initialCapacity = MAXIMUM_CAPACITY;
        int c = initialCapacity / ssize; //  16/15=1
        if (c * ssize < initialCapacity)
            ++c;
        int cap = MIN_SEGMENT_TABLE_CAPACITY; // 2
        while (cap < c)
            cap <<= 1;
        // create segments and segments[0]
        // loadFactor=0.75 
        // cap=2
        // (int)(cap * loadFactor =1
        // 也就是每个分段的加载因子0.75，每个分段里的hash表长度为2，
        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);
    }

Segment

段，是ReentrantLock的子类(只是为了投机取巧，简化一些锁定的操作和避免单独构造)，也叫分段锁，是实现并发的关键。
段维护一个entry 列表，该表始终保持一致状态，因此可以在没有锁定的情况下读取（通过段和表的易失性读取）。
这需要在表调整大小期间在必要时复制节点，因此仍然使用旧版本表的读者可以遍历旧列表。
此类仅定义需要锁定的可变方法。除非另有说明，否则此类的方法执行 ConcurrentHashMap 方法的每段版本。（其他方法直接集成到 ConcurrentHashMap 方法中。）
这些可变方法使用一种通过方法 scanAndLock 和 scanAndLockForPut控制争用旋转的形式。
这些散布在tryLocks中的遍历来定位节点。主要好处是在获取锁的同时吸收缓存未命中（这对于哈希表来说很常见），因此一旦获取，遍历速度就会更快。
我们实际上并没有使用找到的节点，因为无论如何都必须在锁定状态下重新获取它们以确保更新的顺序一致性（并且在任何情况下都可能无法检测到陈旧），但它们通常会更快地重新定位。此外，如果没有找到节点，scanAndLockForPut 会推测性地创建一个新节点以在 put 中使用。

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

        // 每段中的表
        transient volatile HashEntry<K,V>[] table;

        // 元素的个数
        transient int count;

        // 此段中的变异操作总数
        transient int modCount;

        // 当表的大小超过此阈值时，表会重新散列。 （此字段的值始终为(int)(capacity * loadFactor) 。）
        transient int threshold;

        // 加载因子
        final float loadFactor;

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

put操作流程

public V put(K key, V value) {
        Segment<K,V> s;
        if (value == null) 
            throw new NullPointerException();
        int hash = hash(key);
        int j = (hash >>> segmentShift) & segmentMask; // 通过右移去掉低位与掩码求得分段锁的下标
        // 因为没有用volatile修饰，内存不可见的，所以这里在获取分段锁时需要进行复查
        if ((s = (Segment<K,V>)UNSAFE.getObject          
             (segments, (j << SSHIFT) + SBASE)) == null) 
            s = ensureSegment(j); // 确认分段锁
        return s.put(key, hash, value, false); // 将值添加到对应的分段里
    }

    // 通过索引返回对应的分段锁，如果不存在则创建
    private Segment<K,V> ensureSegment(int k) {
        final Segment<K,V>[] ss = this.segments;
        long u = (k << SSHIFT) + SBASE; // raw offset
        Segment<K,V> seg;
        if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) { // 直接从内存里取最新的值
            Segment<K,V> proto = ss[0]; // 用一开始初始化的Segment作为模板进行创建
            int cap = proto.table.length;
            float lf = proto.loadFactor;
            int threshold = (int)(cap * lf);
            HashEntry<K,V>[] tab = (HashEntry<K,V>[])new HashEntry[cap];
            if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
                == null) { // 再次检查
                Segment<K,V> s = new Segment<K,V>(lf, threshold, tab); // 创建新的Segment
                while ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
                       == null) { // 通过CAS操作进行赋值
                    if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s))
                        break;
                }
            }
        }
        return seg;
    }

Segment put操作

核心就是tryLock 和 scanAndLockForPut

final V put(K key, int hash, V value, boolean onlyIfAbsent) {
            HashEntry<K,V> node = tryLock() ? null :
                scanAndLockForPut(key, hash, value); // 先尝试加锁一次，如果加锁失败，再重新尝试循环加锁并扫描给定的key
            V oldValue; // 用来保存旧值
            try {
                HashEntry<K,V>[] tab = table; // 这个table就是分段里的table
                int index = (tab.length - 1) & hash; // 计算索引位置
                HashEntry<K,V> first = entryAt(tab, index); // 根据索引获取链表的首节点
                for (HashEntry<K,V> e = first;;) { // 遍历链表
                    // 1：当前节点不为空
                    if (e != null) { 
                        K k;
                        if ((k = e.key) == key ||
                            (e.hash == hash && key.equals(k))) { // 找到相同的key
                            oldValue = e.value;
                            if (!onlyIfAbsent) {
                                e.value = value; //更新value
                                ++modCount;
                            }
                            break;
                        }
                        e = e.next;
                    }
                    // 2：链表遍历结束也没找到相同key
                    else {
                        if (node != null) // scanAndLockForPut方法中第一次没有获取到锁，遍历链表时不存在相同key,创建新的节点，此时将新节点设置为链表头，“头插法”
                            node.setNext(first);
                        else 
                            // 这里有两种情况：1. scanAndLockForPut方法中第一次获取到了锁；2. 第一次没有获取到锁，遍历链表时存在相同key，此时直接插入，也是“头插法”
                            // 第2中情况说明有其他线程将已经存在的key删除了
                            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;
        }

        // 尝试加锁并扫描给定的key
        // 如果一直获取不到锁，并且未找到key，则创建一个新的节点，否则返回null
        // 循环获取锁，最多尝试64次
        // 返回null有两种情况：1.存在相同key；2.第一次就获取到了锁
        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) { // 只遍历一次链表，key存在的话，就新建节点
                    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) { // MAX_SCAN_RETRIES=64
                    lock();  // 尝试了最大次数之后还是没有获取到锁，采用互斥锁
                    break;
                }
                else if ((retries & 1) == 0 &&
                         (f = entryForHash(this, hash)) != first) {
                    e = first = f;   // (可能有其他线程对遍历的链表进行了修改)每偶数次判断一下是否有改变，否则重新遍历
                    retries = -1;
                }
            }
            return node;
        }

        // 新创建一个2倍大小的数组
        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;
            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;
        }

put 流程总结

remove 操作

public V remove(Object key) {
        int hash = hash(key); // 计算hash值
        Segment<K,V> s = segmentForHash(hash); // 根据hash值找到对应的Segment
        return s == null ? null : s.remove(key, hash, null); // 从Segment中删除
    }

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

Segment remove操作

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))) { // key存在
                        V v = e.value;
                        if (value == null || value == v || value.equals(v)) { // value==null
                            if (pred == null) // 说明key所在节点是链表的首节点
                                setEntryAt(tab, index, next); // 更新首节点为next
                            else
                                pred.setNext(next); // 将上一个节点指向当前节点的下一个节点（删除当前节点）
                            ++modCount;
                            --count;
                            oldValue = v;
                        }
                        break;
                    }
                    pred = e;
                    e = next;
                }
            } finally {
                unlock(); // 释放锁
            }
            return oldValue;
        }

        // 跟scanAndLockForPut类似，但是比它简单一点
        // 不需要考虑新建节点，只需要锁定资源
        private void scanAndLock(Object key, int hash) {
            HashEntry<K,V> first = entryForHash(this, hash);
            HashEntry<K,V> e = first;
            int retries = -1;
            while (!tryLock()) { // 加锁失败
                HashEntry<K,V> f;
                if (retries < 0) { // 第一次加锁失败时遍历链表查找key
                    if (e == null || key.equals(e.key))
                        retries = 0;
                    else
                        e = e.next;
                }
                else if (++retries > MAX_SCAN_RETRIES) { // 最多尝试64次
                    lock();
                    break;
                }
                else if ((retries & 1) == 0 &&
                         (f = entryForHash(this, hash)) != first) { // 链表有改变就重新开始
                    e = first = f;
                    retries = -1;
                }
            }
        }

文章推荐： JavaEE进阶 - Spring 核心与设计思想 - 细节狂魔

文章推荐： MRS离线数据分析：通过Flink作业处理OBS数据

文章推荐： Qt笔记-Centos7搭建OpenSSL开发环境及Qt调用OpenSSL实例

java - 循环遍历 ConcurrentHashMap 的 ConcurrentHashMap
我最初构造了一系列嵌套的 ConcurrentHashMaps private final ConcurrentHashMap allInOne = new ConcurrentHashMap
java - 新 ConcurrentHashMap 新 ConcurrentHashMap
我正在尝试使用 ConcurrentHashMap 初始化 ConcurrentHashMap private final ConcurrentHashMap > myMulitiConcurrent
java - 访问可以是 ConcurrentHashMap 的对象的 ConcurrentHashMap
为了提高工作效率，我尝试将数据保存在一个动态容器中。我在 class 中初始化它与 private final ConcurrentHashMap allInOne = new Concur
java - 我必须将类扩展到 ConcurrentHashMap 还是我可以为 threadSafety 使用变量 ConcurrentHashMap
我正在创建基于 Socket 的服务器-客户端预订服务，并且遇到有关将由多个线程访问的类的问题，是否需要扩展 ConcurrentHashMap 或创建变量 ConcurrentHashMap 是否足
java - ConcurrentHashMap.put V.S. ConcurrentHashMap.replace
从 Javadoc 我知道 ConcurrentHashMap.replace 是原子的，但是 ConcurrentHashMap.put 呢？我看到它们在源代码中的实现方式不同，但我无法弄清楚它们的
java - 是否保证 ConcurrentHashMap.get() 可以通过不同的线程看到以前的 ConcurrentHashMap.put() ？
是 ConcurrentHashMap.get() 保证看到以前的ConcurrentHashMap.put()通过不同的线程？我的期望是，阅读 JavaDocs 似乎表明了这一点，但我 99% 确信
java - 为什么 ConcurrentHashMap::putIfAbsent 比 ConcurrentHashMap::computeIfAbsent 快？
使用 ConcurrentHashMap，我发现 computeIfAbsent 比 putIfAbsent 慢两倍。这里是简单的测试: import java.util.ArrayList; imp
java - 如何从此 ConcurrentHashMap> 中删除一个值
我有一个以下格式的 ConcurrentHashMap: ConcurrentHashMap> 现在在此 map 中，我想删除数组列表中的特定值。任何人都可以指导这一点。编辑1:我有一张 map >
java - 为什么 ConcurrentHashMap.Segment 和 ConcurrentHashMap.HashEntry 类是静态的？
为什么 ConcurrentHashMap.Segment 和 ConcurrentHashMap.HashEntry 类是静态的？为什么要这样设计？最佳答案基本上所有不需要使用其封闭类属性的内部
scala - ConcurrentHashMap[String, AtomicInteger] 或 ConcurrentHashMap[String, Int] 用于线程安全计数器？
在 ConcurrentHashMap 中通过键递增并发计数器时，使用常规 Int 作为值是否安全，还是我们必须使用 AtomicInteger？例如考虑以下两个实现 ConcurrentHashMa
java ConcurrentHashMap of Integer vs HashMap of AtomicInteger vs ConcurrentHashMap of AtomicInteger
我对java中的并发数据结构有疑问，特别是: 1) ConcurrentHashMap 2) HashMap 3) ConcurrentHashMap 如果我理解正确的话: 1) 读/写是线程安全的，
java - 在 ConcurrentHashMap.computeIfAbsent 和 ConcurrentHashMap.computeIfPresent 中执行 `mappingFunction`
我正在尝试查看实际的 Java 文档，描述传递给 ConcurrentHashMap.computeIfAbsent 和 ConcurrentHashMap.computeIfPresent< 时可以
java - 当 B 实现 A 时，将 ConcurrentHashMap 转换为 ConcurrentHashMap
我有一个名为 SerializableL 的接口(interface)由 3 个不同的类实现: 产品横幅标签我开始重构，想用多个方法调用替换多个段落。 public void load(Conc
ConcurrentHashMap 的底层结构与演进过程
一 JDK 中的 ConcurrentHashMap 在 JDK 8以前，HashMap 是基于数组 + 链表来实现的。整体上看，HashMap 是一个数组，但每个数组元素又是一张链表。当向 Has
调整大小时读取 ConcurrentHashMap
我想知道当我们在调整大小时尝试读取 ConcurrentHashMap 时可能发生的情况。我知道在读取期间，第一次尝试总是不同步的。在第二次尝试中，它将尝试获取锁并重试。但是，如果它在调整大小时发
Java ConcurrentHashMap
在一个应用程序中，1 个线程负责不断更新映射，主线程定期读取映射，使用 ConcurrentHashmap 是否足够？或者我应该明确地锁定同步块(synchronized block)中的操作吗？任何
Java集合-ConcurrentHashMap
介绍 ConcurrentHashMap 技术是为了解决问题而生的，ConcurrentHashMap 解决了多个线程同时操作一个 HashMap 时，可能出现的内部问题。当多个线程同时操作一
java - ConcurrentHashMap 上的同步是否正确？
我有一个由多个线程访问的键值映射: private final ConcurrentMap key_vval_map = new ConcurrentHashMap(); 我的自定义 get() 和
java - ConcurrentHashMap - 奇怪的行为
谁能告诉我这段代码出了什么问题？我要拔头发了! 如果我使用 HashMap 而不是 ConcurrentHashMap 则没有任何问题。代码使用JDK 5.0编译 public class MapTe
java - ConcurrentHashMap 中的锁定和锁定场景之前的重试
来自 ConcurrentHashMap 的源码 /** 171 * Number of unsynchronized retries in size and containsVal

知者

个人简介

我是一名优秀的程序员,十分优秀！

作者热门文章

滴滴打车优惠券免费领取

全站热门文章

首页

博学

6Ren·AI

商城