ConcurrentHashMap（1.7版本）

Posted 2022-09-23 lovezmc

1、基础属性

static final int DEFAULT_INITIAL_CAPACITY = 16;//默认初始容量
static final float DEFAULT_LOAD_FACTOR = 0.75f;//默认加载因子
static final int DEFAULT_CONCURRENCY_LEVEL = 16;//默认线程并发度，默认最大允许16个线程并发访问，条件是这16个线程在分别在16个segments中
static final int MAXIMUM_CAPACITY = 1 << 30;//最大容量
static final int MIN_SEGMENT_TABLE_CAPACITY = 2;//每个segment的最小容量
static final int MAX_SEGMENTS = 1 << 16;//segment个数限制
static final int RETRIES_BEFORE_LOCK = 2; // 计算size时使用
final int segmentMask;// 用于定位segment，大小等于segments数组的大小减1
final int segmentShift;// 用于定位段，大小等于32(hash值的位数)减去对segments的大小取以2为底的对数值
final Segment<K,V>[] segments;//存放数据的载体
transient Set<K> keySet;
transient Set<Map.Entry<K,V>> entrySet;
transient Collection<V> values;

2、Segment结构

static final class Segment<K,V> extends ReentrantLock implements Serializable 
    private static final long serialVersionUID = 2249069246763182397L;

    /**
     * The maximum number of times to tryLock in a prescan before
     * possibly blocking on acquire in preparation for a locked
     * segment operation. On multiprocessors, using a bounded
     * number of retries maintains cache acquired while locating
     * nodes.
     */
    static final int MAX_SCAN_RETRIES =
        Runtime.getRuntime().availableProcessors() > 1 ? 64 : 1;

    transient volatile HashEntry<K,V>[] table;//每个Segment的数据都是一个 "类hashmap"

    transient int count;//当前Segment的元素个数
    
    transient int modCount;

    transient int threshold;// 扩容阈值

    final float loadFactor;//加载因子, threshold=capacity*loadFactor

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

    final V put(K key, int hash, V value, boolean onlyIfAbsent) 
        // 尝试获取锁，如果获取不到就进入到scanAndLockForPut方法去持续获取锁(获取不到锁就不干了)
        HashEntry<K,V> node = tryLock() ? null : scanAndLockForPut(key, hash, value);
        V oldValue;
        try 
            HashEntry<K,V>[] tab = table;
            int index = (tab.length - 1) & hash;
            HashEntry<K,V> first = entryAt(tab, index);//获取链表头
            for (HashEntry<K,V> e = first;;) 
                if (e != null) 
                    // 先判断该key值是否已存在，若存在则替换
                    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;
    

    /**
     * Doubles size of table and repacks entries, also adding the
     * given node to new table
     */
    @SuppressWarnings("unchecked")
    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];
        /**
          * 掩码是length-1，而length又是2的幂, 即掩码为00...0011...11(前面全0，后面全1)
          * 寻址方式  idx = hash & sizeMask
          * 优点：hash由key得来，是固定的，那么扩容之后，新的idx的值只可能是两个，我们在重构的时候，总有那么一段链表是可以看作一个整体放在数组的某个位置的
          * ex：    old：容量16，掩码1111；  new：容量32，掩码11111
          * 由于hash值是不变的，与运算后低四位与原来不变，只有第五位才会变(0或1)，若为0，则newIdx不变，若为1，则newIdx = oldLength+oldIdx
          */
        int sizeMask = newCapacity - 1;
        // 接下来就是重构这个segment了
        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)   //  只有这么一个节点，直接放就好了
                    newTable[idx] = e;
                else 
                    HashEntry<K,V> lastRun = e;
                    int lastIdx = idx;
                    // lastRun特性：在这个链表中，lastRun后面的所有node的 idx 与 lastRun的 idx 一致，这样我们就可以将lastRun及其后面的所有node看作一个整体
                    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;
                    // 处理lastRun之前的每一个节点
                    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;
    

    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;
            
            // 重试次数超过限制时，则阻塞至获取到锁(lock()时阻塞方法，获取到锁后返回)
            else if (++retries > MAX_SCAN_RETRIES) 
                lock();
                break;
            
            else if ((retries & 1) == 0 &&
                    //如果有新的元素进入链表成为了表头，则重新走一遍该方法
                     (f = entryForHash(this, hash)) != first) 
                e = first = f; // re-traverse if entry changed
                retries = -1;
            
        
        return node;
    

    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;
            
        
    

    /**
     * Remove; match on key only if value null, else match both.
     */
    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;
    

    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;
    

    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、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;
    

4、构造函数

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;    // 用于定位segment，大小为 lg(ssize)
    int ssize = 1;    // segments的大小，即总的segment个数
    while (ssize < concurrencyLevel) 
        ++sshift;
        ssize <<= 1;
    
    this.segmentShift = 32 - sshift;  // 32-lg(ssize)
    this.segmentMask = ssize - 1;  // 掩码，用以计算segments的下标
    if (initialCapacity > MAXIMUM_CAPACITY)
        initialCapacity = MAXIMUM_CAPACITY;
    int c = initialCapacity / ssize; // 总容量 / segment数量 = 每个segment的capacity
    if (c * ssize < initialCapacity) // 进一法
        ++c;
    int cap = MIN_SEGMENT_TABLE_CAPACITY;
    while (cap < c)
        cap <<= 1;
    // 创建segments并初始化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;

5、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; // segments寻址
    if ((s = (Segment<K,V>)UNSAFE.getObject (segments, (j << SSHIFT) + SBASE)) == null) // 判断是否需要初始化segments[j]
        s = ensureSegment(j);// 因为构造方法中只初始化了segments[0]，其它位置需要初始化
    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]; // 这里相当于拷贝了一份segments[0]的架构
        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);
            // 使用 while 循环，判断该槽是否已被其它线程初始化，如果没有，内部用 CAS，当前线程成功设值或其他线程成功设值后，退出；否则每次都要判断一次
            while ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) 
                if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s)) // CAS
                    break;
            
        
    
    return seg;

6、size计算：在不加锁的情况下先后计算两次，如果两次计算的modCount一致，则认为在统计的时间内，没有其它线程对该map修改或删除，直接返回size；如果两次计算的modCount不一致，则对所有的Segment加锁，并计算size

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)  // 计算了两次的modcount不一致, RETRIES_BEFORE_LOCK=2
                for (int j = 0; j < segments.length; ++j)
                    ensureSegment(j).lock(); // 对每个segment加锁
            
            sum = 0L; // 统计modCount
            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;