Java基础-ConcurrentHashMap

Java 7 中 ConcurrentHashMap 的存储结构如下图，ConcurrnetHashMap 由很多个 Segment 组合，而每一个 Segment 内部是一个类似于 HashMap 结构的 HashEntry 数组，所以可以进行扩容。但是 Segment 的个数一旦初始化就不能改变，默认 Segment 的个数是 16 个，可以认为 ConcurrentHashMap 默认支持最多 16 个线程并发。

一个ConcurrentHashMap中只有一个Segment类型的segments数组，每个segment中只有一个HashEntry类型的table数组，table数组中存放一个HashEntry节点。

    
    static final int DEFAULT_INITIAL_CAPACITY = 16;

    
    static final float DEFAULT_LOAD_FACTOR = 0.75f;

    
    static final int DEFAULT_CONCURRENCY_LEVEL = 16;

1. capacity：与小数组（HashEntry）的容量有关（小数组的初始容量 = capacity / clevel，最小是2）
2. factor：负载因子，影响的是小数组的扩容
3. clevel：并发度，也即Segment 的个数，大数组的个数，默认是16

@SuppressWarnings("unchecked")
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;
    // c = 容量 / ssize ，默认 16 / 16 = 1，这里是计算HashEntry的容量
    int c = initialCapacity / ssize;
    if (c * ssize < initialCapacity)
        ++c;
    int cap = MIN_SEGMENT_TABLE_CAPACITY;
    //HashEntry的容量至少是2或者2的倍数
    while (cap < c)
        cap <<= 1;
    // create segments and segments[0]
    // 创建 Segment 数组，设置 segments[0]
    Segment s0 = new Segment(loadFactor, (int)(cap * loadFactor),
                         (HashEntry[])new HashEntry[cap]);
    Segment[] ss = (Segment[])new Segment[ssize];
    UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]
    this.segments = ss;
}

构造函数中的重点：

1. 记录偏移量sshift=clevel 2的N次方中N，得到segmentShift = 32 - sshift;用来确定目标值在Segment中的位置
2. 记录segmentMask = ssize - 1;使得segmentMask的每一位都为1
3. Segment的容量ssize = 大于等于clevel的最小2次幂
4. HashEntry的容量cap = 大于等于 ( 初始容量 / ssize ) 的最小2次幂

public V put(K key, V value) {
    Segment s;
    if (value == null)
        throw new NullPointerException();
    int hash = hash(key);
    int j = (hash >>> segmentShift) & segmentMask;
    if ((s = (Segment)UNSAFE.getObject          // nonvolatile; recheck
         (segments, (j << SSHIFT) + SBASE)) == null) //  in ensureSegment
        // 如果查找到的 Segment 为空，初始化
        s = ensureSegment(j);
    return s.put(key, hash, value, false);
}

@SuppressWarnings("unchecked")
private Segment ensureSegment(int k) {
    final Segment[] ss = this.segments;
    long u = (k << SSHIFT) + SBASE; // raw offset
    Segment seg;
    // 判断 u 位置的 Segment 是否为null
    if ((seg = (Segment)UNSAFE.getObjectVolatile(ss, u)) == null) {
        Segment proto = ss[0]; // use segment 0 as prototype
        // 获取0号 segment 里的 HashEntry 初始化长度
        int cap = proto.table.length;
        // 获取0号 segment 里的 hash 表里的扩容负载因子，所有的 segment 的 loadFactor 是相同的
        float lf = proto.loadFactor;
        // 计算扩容阀值
        int threshold = (int)(cap * lf);
        // 创建一个 cap 容量的 HashEntry 数组
        HashEntry[] tab = (HashEntry[])new HashEntry[cap];
        if ((seg = (Segment)UNSAFE.getObjectVolatile(ss, u)) == null) { // recheck
            // 再次检查 u 位置的 Segment 是否为null，因为这时可能有其他线程进行了操作
            Segment s = new Segment(lf, threshold, tab);
            // 自旋检查 u 位置的 Segment 是否为null
            while ((seg = (Segment)UNSAFE.getObjectVolatile(ss, u))
                   == null) {
                // 使用CAS 赋值，只会成功一次
                if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s))
                    break;
            }
        }
    }
    return seg;
}

流程如下：

1. 根据segmentShift和segmentMask确定要插入的元素在segment中的位置
2. 如果查找到的segment位置上为空，则进入初始化
3. 根据segment[0]里HashEntry的容量和负载因子，计算阈值，创建新的HashEntry数组
4. 判断segment中该位置上是否仍为null，如果仍然是则使用CAS赋值

将得到的hash值向右按位移动segmentShift位，然后再与segmentMask做&运算得到Segment的索引。
在初始化的时候，segmentShift的值等于32-sshift，例如concurrencyLevel等于16，则sshift等于4，那么segmentShift为28。hash值是一个32位的整数，将其向右移动28就变成这个样子：0000 0000 0000 0000 0000 0000 0000 XXXX，然后再用这个值和segmentMask做与运算，所得结果确定Segment的索引。

以上仅仅是初始化segment的操作，在初始化完成后将进入插入操作：

final V put(K key, int hash, V value, boolean onlyIfAbsent) {
    //先尝试对segment加锁，如果直接加锁成功，那么node=null；如果加锁失败，则会调用scanAndLockForPut方法去获取锁，
    //在这个方法中，获取锁后会返回对应HashEntry（要么原来就有要么新建一个）
    HashEntry node = tryLock() ? null :
        scanAndLockForPut(key, hash, value);
    V oldValue;
    try {
        //这里是一个优化点，由于table自身是被volatile修饰的，然而put这一块代码本身是加锁了的，所以同一时间内只会有一个线程操作这部分内容，
        //所以不再需要对这一块内的变量做任何volatile修饰，因为变量加了volatile修饰后，变量无法进行编译优化等，会对性能有一定的影响
        //故将table赋值给put方法中的一个局部变量，从而使得能够减少volatile带来的不必要消耗。
        
        HashEntry[] tab = table;
        int index = (tab.length - 1) & hash;
        
        //这里有一个问题：为什么不直接使用数组下标获取HashEntry，而要用entryAt来获取链表？
        //这里结合网上内容个人理解是：由于Segment继承的是ReentrantLock，所以它是一个可重入锁，那么是否存在某种场景下，
        //会导致同一个线程连续两次进入put方法，而由于put最终使用的putOrderedObject只是禁止了写写重排序无法保证内存可见性，
        //所以这种情况下第二次put在获取链表时必须用entryAt中的volatile语义的get来获取链表，因为这种情况下下标获取的不一定是最新数据。
       
        HashEntry first = entryAt(tab, index);//先获取需要put的对在当前这个segment中对应的链表的表头结点。

        for (HashEntry e = first;;) {//开始遍历first为头结点的链表
            if (e != null) {//<1>
                //e不为空，说明当前键值对需要存储的位置有hash冲突，直接遍历当前链表，如果链表中找到一个节点对应的key相同，
                //依据onlyIfAbsent来判断是否覆盖已有的value值
                K k;
                if ((k = e.key) == key ||
                    (e.hash == hash && key.equals(k))) {  //说明键的Hash值一样，内容也一样
                    //进入这个条件内说明需要put的对应的key节点已经存在，直接判断是否更新并最后break退出循环。
                    oldValue = e.value;
                    if (!onlyIfAbsent) {
                        e.value = value;
                        ++modCount;
                    }
                    break;
                }
                e = e.next;//未进入上面的if条件中，说明当前e节点对应的key不是需要的，直接遍历下一个节点。
            }
            else {//<2> 头插法
                //进入到这个else分支，说明e为空，对应有两种情况下e可能会为空，即：
                // 1>. <1>中进行循环遍历，遍历到了链表的表尾仍然没有满足条件的节点。
                // 2>. e=first一开始就是null（可以理解为即一开始就遍历到了尾节点）
                
                if (node != null) //这里有可能获取到锁是通过scanAndLockForPut方法内自旋获取到的，这种情况下依据找好或者说是新建好了对应节点，node不为空
                    node.setNext(first);
                else     // 当然也有可能是这里直接第一次tryLock就获取到了锁，从而node没有分配对应节点，即需要给依据插入的k,v来创建一个新节点
                    node = new HashEntry(hash, key, value, first);
                int c = count + 1; //总数+1 在这里依据获取到了锁，即是线程安全的！对应了上述对count变量的使用规范说明。
                if (c > threshold && tab.length < MAXIMUM_CAPACITY)//判断是否需要进行扩容
                
                    //扩容是直接重新new一个新的HashEntry数组，这个数组的容量是老数组的两倍，
                    //新数组创建好后再依次将老的table中的HashEntry插入新数组中，所以这个过程是十分费时的，应尽量避免。
                    //扩容完毕后，还会将这个node插入到新的数组中。
                    
                    rehash(node);
                else
                    setEntryAt(tab, index, node);
                ++modCount;
                count = c;
                oldValue = null;
                break;
            }
        }
    } finally {
        unlock();
    }
    return oldValue;
}

插入过程如下：

1. 计算得到插入元素的目标位置，获取这个位置上的HashEntry元素。
2. 如果这个位置上的头结点first!=null，遍历链表，找key值相同或者key值和value值都相同的结点，根据onlyIfAbsent判断是否覆盖oldValue
3. 如果这个位置上的头结点first==null，有两种可能：1⃣️遍历到链表尾部都没有一样的结点2⃣️fist==null。使用头插法将目标结点插入到链表中，如果插入后的容量大于阈值，则扩容。

private void rehash(HashEntry node) {
    HashEntry[] oldTable = table;
    // 老容量
    int oldCapacity = oldTable.length;
    // 新容量，扩大两倍
    int newCapacity = oldCapacity << 1;
    // 新的扩容阀值 
    threshold = (int)(newCapacity * loadFactor);
    // 创建新的数组
    HashEntry[] newTable = (HashEntry[]) new HashEntry[newCapacity];
    // 新的掩码，默认2扩容后是4，-1是3，二进制就是11。
    int sizeMask = newCapacity - 1;
    for (int i = 0; i < oldCapacity ; i++) {
        // 遍历老数组
        HashEntry e = oldTable[i];
        if (e != null) {
            HashEntry 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 lastRun = e;
                int lastIdx = idx;
                // 新的位置只可能是不便或者是老的位置+老的容量。
                // 遍历结束后，lastRun 后面的元素位置都是相同的
                for (HashEntry last = next; last != null; last = last.next) {
                    int k = last.hash & sizeMask;
                    if (k != lastIdx) {
                        lastIdx = k;
                        lastRun = last;
                    }
                }
                // lastRun 后面的元素位置都是相同的，直接作为链表赋值到新位置。
                newTable[lastIdx] = lastRun;
                // Clone remaining nodes
                for (HashEntry p = e; p != lastRun; p = p.next) {
                    // 遍历剩余元素，头插法到指定 k 位置。
                    V v = p.value;
                    int h = p.hash;
                    int k = h & sizeMask;
                    HashEntry n = newTable[k];
                    newTable[k] = new HashEntry(h, p.key, v, n);
                }
            }
        }
    }
    // 头插法插入新的节点
    int nodeIndex = node.hash & sizeMask; // add the new node
    node.setNext(newTable[nodeIndex]);
    newTable[nodeIndex] = node;
    table = newTable;
}

扩容的流程：

1. 新的HashEntry[]容量是原来的二倍，负载因子不变，可以求出新的阈值，以及新的sizeMask，根据新的sizeMask和新的容量可以定位元素在新的数组中的位置
2. 遍历原来的数组，计算数组中元素新的hash值，定位到新的数组中，判断该元素是否有下一个结点
3. 如果没有下个结点，可以直接赋值到新的数组
4. 如果有下个结点，进入到第一个for循环中，目的是找到链表后面几个hash值相同并相连的结点，如果找到可以直接批量赋值到新的数组
5. 然后进入到第二个for循环中，把剩下的结点再放在新的数组中
6. 最后把输入的结点插入到新的数组中 

    public V get(Object key) {
        Segment s; // manually integrate access methods to reduce overhead
        HashEntry[] tab;
        //1 和put操作一样，先通过key进行两次hash确定取哪个segment中的数据
        int h = hash(key);
        //2 使用UNSAFE方法获取对应的Segment，然后再进行一次&运算得到HashEntry链表的位置，然后从链表头开始遍历整个链表。
        //（由于hash会碰撞，所以用一个链表保存），如果找到对应的key，则返回对应的value值，如果链表遍历完都没有找到对应的key，
        // 则说明map中不包含该key，返回null
        long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
        if ((s = (Segment)UNSAFE.getObjectVolatile(segments, u)) != null &&
            (tab = s.table) != null) {
            for (HashEntry e = (HashEntry) UNSAFE.getObjectVolatile
                     (tab, ((long)(((tab.length - 1) & h)) << TSHIFT) + TBASE);
                 e != null; e = e.next) {
                K k;
                // 如果发生哈希碰撞，判断key是否相同
                if ((k = e.key) == key || (e.hash == h && key.equals(k)))
                    return e.value;
            }
        }
        return null;
    }

查找的流程如下：

1. 求得目标元素在segment中的位置，得到该位置上的HashEntry数组
2. 求得目标元素在HashEntry中的位置，遍历链表找到key相同的元素，返回value值

https://javaguide.cn/java/collection/concurrent-hash-map-source-code.html#_1-concurrenthashmap-1-7 

https://blog.csdn.net/fuyuwei2015/article/details/72630365