ConcurrentHashMap源码解析_02 预热(内部一些小方法分析)

我就是我 2022-10-21 11:55 58阅读 0赞

前面一篇文章中介绍了并发HashMap的主要成员属性，内部类和构造函数，下面在正式分析并发HashMap成员方法之前，先分析一些内部类中的字方法函数：

首先来看下ConcurrentHashMap内部类Node中的hash成员属性值的计算方法`spread(int h)`：

static class Node<K,V> implements Map.Entry<K,V> { 
        final int hash;// 该属性是通过spread(int h)方法计算得到~
        final K key;
        volatile V val;
        volatile Node<K,V> next;
        Node(int hash, K key, V val, Node<K,V> next) { 
            this.hash = hash;
            this.key = key;
            this.val = val;
            this.next = next;
        }
        ...
    }

## 1、spread(int h)方法 ##

/** * 计算Node节点hash值的算法：参数h为hash值 * eg: * h二进制为 --> 1100 0011 1010 0101 0001 1100 0001 1110 * (h >>> 16) --> 0000 0000 0000 0000 1100 0011 1010 0101 * (h ^ (h >>> 16)) --> 1100 0011 1010 0101 1101 1111 1011 1011 * 注：(h ^ (h >>> 16)) 目的是让h的高16位也参与寻址计算，使得到的hash值更分散，减少hash冲突产生~ * ------------------------------------------------------------------------------ * HASH_BITS --> 0111 1111 1111 1111 1111 1111 1111 1111 * (h ^ (h >>> 16)) --> 1100 0011 1010 0101 1101 1111 1011 1011 * (h ^ (h >>> 16)) & HASH_BITS --> 0100 0011 1010 0101 1101 1111 1011 1011 * 注： (h ^ (h >>> 16))得到的结果再& HASH_BITS，目的是为了让得到的hash值结果始终是一个正数 */
    static final int spread(int h) { 
        // 让原来的hash值异或^原来hash值的右移16位，再与&上HASH_BITS(0x7fffffff: 二进制为31个1)
        return (h ^ (h >>> 16)) & HASH_BITS;
    }

下面介绍`tabAt(Node<K,V>[] tab, int i)`方法：获取 tab(Node\[\]) 数组指定下标 i 的Node节点。

## 2、tabAt方法 ##

/** * 获取 tab(Node[]) 数组指定下标 i 的Node节点 * Node<K,V>[] tab：表示Node[]数组 * int i：表示数组下标 */
    static final <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i) { 
        // ((long)i << ASHIFT) + ABASE 的作用：请看下面的分析描述~
        return (Node<K,V>)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE);
    }

在分析`((long)i << ASHIFT) + ABASE`时，先复习一下上一篇文章：[ConcurrentHashMap源码解析\_01 成员属性、内部类、构造方法分析][ConcurrentHashMap_01]中介绍到的一些属性字段的含义：

// Node数组的class对象
    Class<?> ak = Node[].class;
    
    // U.arrayBaseOffset(ak)：根据as获取Node[]数组第一个元素的偏移地址ABASE
    ABASE = U.arrayBaseOffset(ak);
    
    // scale：表示数组中每一个单元所占用的空间大小，即scale表示Node[]数组中每一个单元所占用的空间
    int scale = U.arrayIndexScale(ak);// scale必须为2的次幂数
    
    // numberOfLeadingZeros(scale) 根据scale，返回当前数值转换为二进制后，从高位到地位开始统计，统计有多少个0连续在一块：eg, 8转换二进制=>1000 则 numberOfLeadingZeros(8)的结果就是28，为什么呢？因为Integer是32位，1000占4位，那么前面就有32-4个0，即连续最长的0的个数为28个
    // 4转换二进制=>100 则 numberOfLeadingZeros(8)的结果就是29
    // ASHIFT = 31 - Integer.numberOfLeadingZeros(4) = 2 那么ASHIFT的作用是什么呢？其实它有数组寻址的一个作用：
    // 拿到下标为5的Node[]数组元素的偏移地址(存储地址)：假设此时 根据scale计算得到的ASHIFT = 2
    // ABASE + (5 << ASHIFT) == ABASE + (5 << 2) == ABASE + 5 * scale(乘法运算效率低)，就得到了下标为5的数组元素的偏移地址
    ASHIFT = 31 - Integer.numberOfLeadingZeros(scale);

由上面的几个属性字段的复习介绍，不难得出：

*  `((long)i << ASHIFT) + ABASE` 就是得到当前`Node[]` 数组下标为`i`的节点对象的偏移地址。
 *  然后再通过`(Node<K,V>)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE)` 方法，根据`Node[]`和`目标节点Node的偏移地址`两个参数，得到下标为i的Node节点对象。
 *  虽然这样很绕，不如，但是直接根据偏移地址去寻找数组元素效率较高~

## 3、casTabAt方法 ##

/** * 通过CAS的方式去向Node数组指定位置i设置节点值，设置成功返回true，否则返回false * Node<K,V>[] tab：表示Node[]数组 * int i：表示数组下标 * Node<K,V> c：期望节点值 * Node<K,V> v：要设置的节点值 */
    static final <K,V> boolean casTabAt(Node<K,V>[] tab, int i,
                                        Node<K,V> c, Node<K,V> v) { 
        // 调用Unsafe的比较并交换去设置Node[]数组指定位置的节点值，参数如下：
        // tab：Node[]数组
        // ((long)i << ASHIFT) + ABASE：下标为i数组桶的偏移地址
        // c：期望节点值
        // v：要设置的节点值
        return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v);
    }

## 4、setTabAt方法 ##

/** * 根据数组下标i，设置Node[]数组指定下标位置的节点值： * Node<K,V>[] tab：表示Node[]数组 * int i：表示数组下标 * Node<K,V> v：要设置的节点值 */
    static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) { 
        // ((long)i << ASHIFT) + ABASE：下标为i数组桶的偏移地址
        U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v);
    }

## 5、resizeStamp方法 ##

/** * table数组扩容时，计算出一个扩容标识戳，当需要并发扩容时，当前线程必须拿到扩容标识戳才能参与扩容： */
    static final int resizeStamp(int n) { 
        // RESIZE_STAMP_BITS：固定值16，与扩容相关，计算扩容时会根据该属性值生成一个扩容标识戳
        return Integer.numberOfLeadingZeros(n) | (1 << (RESIZE_STAMP_BITS - 1));
    }

举例子分析一下：

*  当我们需要table容量从16 库容到32时：`Integer.numberOfLeadingZeros(16)` 会得到 `27`，怎么得来的呢？
    
     *  `numberOfLeadingZeros(n)` 根据传入的`n`，返回当前数值转换为二进制后，从高位到地位开始统计，统计有多少个0连续在一块：
        
         *  eg：16 转换二进制 => `1 0000` 则 `numberOfLeadingZeros(16)`的结果就是 `27`，因为`Integer`是32位，`1 0000`占5位，那么前面就有`(32 - 5)`个0，即连续最长的0的个数为27个。
 *  `(1 << (RESIZE_STAMP_BITS - 1))`：其中`RESIZE_STAMP_BITS`是一个固定值16，**与扩容相关，计算扩容时会根据该属性值生成一个扩容标识戳**。
 *  下面就来计算一下：

// 从16扩容到32
    16 -> 32 
    // 用A表示：
    numberOfLeadingZeros(16) => 1 0000 => 27 => 0000 0000 0001 1011
    // 用B表示：
    (1 << (RESIZE_STAMP_BITS - 1)) => (1 << (16 - 1) => 1000 0000 0000 0000 => 32768
    
    // A | B 
    Integer.numberOfLeadingZeros(n) | (1 << (RESIZE_STAMP_BITS - 1)) 
    -----------------------------------------------------------------
    0000 0000 0001 1011  ---> A
    1000 0000 0000 0000  ---> B
    -------------------  ---> | 按位或                             
    1000 0000 0001 1011  ---> 计算得到扩容标识戳

## 6、tableSizeFor方法 ##

/** * 根据c,计算得到大于等于c的，最小2的次幂数，该方法在HashMap源码中分析过~ * eg：c = 28 ，则计算得到的返回结果为 32 * c = 28 ==> n=27 => 0b 11011 * * 11011 | 01101 => 11111 ---- n |= n >>> 1 * 11111 | 00111 => 11111 ---- n |= n >>> 2 * .... * => 11111 + 1 = 100000 = 32 */
    private static final int tableSizeFor(int c) { 
        int n = c - 1;
        n |= n >>> 1;
        n |= n >>> 2;
        n |= n >>> 4;
        n |= n >>> 8;
        n |= n >>> 16;
        return (n < 0) ? 1 : (n >= MAXIMUM_CAPACITY) ? MAXIMUM_CAPACITY : n + 1;
    }

## 7、构造方法(复习) ##

复习一下上一篇文章中的并发HashMap的狗构造方法：

// 无惨构造
    public ConcurrentHashMap() { 
    }
    
    // 指定初始化容量
    public ConcurrentHashMap(int initialCapacity) { 
        if (initialCapacity < 0)
            throw new IllegalArgumentException();
        int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?
                MAXIMUM_CAPACITY :
                tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));
        /** * sizeCtl > 0 * 当目前table未初始化时，sizeCtl表示初始化容量 */
        this.sizeCtl = cap;
    }
    
    // 根据一个Map集合来初始化
    public ConcurrentHashMap(Map<? extends K, ? extends V> m) { 
        // sizeCtl设置为默认容量值
        this.sizeCtl = DEFAULT_CAPACITY;
        putAll(m);
    }
    
    // 指定初始化容量，和负载因子
    public ConcurrentHashMap(int initialCapacity, float loadFactor) { 
        this(initialCapacity, loadFactor, 1);
    }
    
    // 指定初始化容量，和负载因子，并发级别
    public ConcurrentHashMap(int initialCapacity,
                             float loadFactor, int concurrencyLevel) { 
        if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0)
            throw new IllegalArgumentException();
        // 当指定的初始化容量initialCapacity小于并发级别concurrencyLevel时
        if (initialCapacity < concurrencyLevel) 
            // 初始化容量值设置为并发级别的值。
            // 即，JDK1.8以后并发级别由散列表长度决定
            initialCapacity = concurrencyLevel; 
        // 根据初始化容量和负载因子，去计算size
        long size = (long)(1.0 + (long)initialCapacity / loadFactor);
        // 根据size重新计算数组初始化容量
        int cap = (size >= (long)MAXIMUM_CAPACITY) ?
                MAXIMUM_CAPACITY : tableSizeFor((int)size);
        /** * sizeCtl > 0 * 当目前table未初始化时，sizeCtl表示初始化容量 */
        this.sizeCtl = cap;
    }

--------------------

预热结束后，下一篇文章就是并发Map比较重点的地方了，即`put()`写入操作原理~

[ConcurrentHashMap_01]: https://csp1999.blog.csdn.net/article/details/116380238