1. ConcurrentHashMap1.7

1. 存储结构

Segment 数组 + HashEntry 数组 + 链表
Segment 数组 + HashEntry 数组 + 链表

2. 初始化方法

初始化逻辑
无参构造
有参构造
  1. 必要参数校验。
  2. 校验并发级别 concurrencyLevel 大小，如果大于最大值，重置为最大值。无参构造默认值是 16.
  3. 寻找并发级别 concurrencyLevel 之上最近的 2 的幂次方值，作为初始化容量大小，记为ssize，默认是 16。
  4. 记录 segmentShift 偏移量，这个值为【容量 = 2 的 N 次方】中的 N(记为：sshift)，默认是 32 - sshift = 28。
    记录 segmentMask为掩码，默认是 ssize - 1 = 16 -1 = 15.
  5. 初始化 segments[0]，默认大小为 2，负载因子 0.75，扩容阀值是 2*0.75=1.5，插入第二个值时才会进行扩容。
  6. 所以：concurrencyLevel=16,ssize=16,sshift=4,this.segmentShift = 32 - sshift=28,而this.segmentMask = ssize - 1=15 二进制就是1111;
  7. （hash >>> segmentShift) & segmentMask //定位Segment所使用的hash算法
    int index = hash & (tab.length - 1); // 定位HashEntry所使用的hash算法
1
2
3
4
5
6
7
8
public ConcurrentHashMap() {
 	/**
	 * 默认初始化容量
	 * 默认负载因子
	 * 默认并发级别
      */
     this(DEFAULT_INITIAL_CAPACITY, DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
@SuppressWarnings("unchecked")
public ConcurrentHashMap(int initialCapacity,float loadFactor, int concurrencyLevel) {
 // 参数校验
 if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0)
     throw new IllegalArgumentException();
 // 校验并发级别大小，大于 1<<16，重置为 65536
 if (concurrencyLevel > MAX_SEGMENTS)
     concurrencyLevel = MAX_SEGMENTS;
 // Find power-of-two sizes best matching arguments
 // 2的多少次方
 int sshift = 0;
 // segment大小
 int ssize = 1;
 // 这个循环可以找到 concurrencyLevel 之上最近的 2的次方值
 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，这里是计算每个 Segment 中的类似于 HashMap 的容量
 int c = initialCapacity / ssize;
 if (c * ssize < initialCapacity)
     ++c;
 int cap = MIN_SEGMENT_TABLE_CAPACITY;
 //Segment 中的类似于 HashMap 的容量至少是2或者2的倍数
 while (cap < c)
     cap <<= 1;
 // create segments and segments[0]
 // 创建 Segment 数组，设置 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;
}

3. put()方法

原理
code-put()
code-ensureSegment()-初始化segment
code-put()
code-scanAndLockForPut()-不断自旋获取锁
  1. 计算要 put 的 key 的位置，获取指定位置的 Segment.
  2. 如果指定位置的 Segment 为空，则初始化这个 Segment.
  3. Segment.put 插入 key,value 值.
1
2
3
4
5
6
7
8
9
10
11
12
13
14
public V put(K key, V value) {
    Segment<K,V> s;
    if (value == null)
        throw new NullPointerException();
    int hash = hash(key);
    // hash 值无符号右移 28位（初始化时获得），然后与 segmentMask=15 做与运算
    // 其实也就是把高4位与segmentMask（1111）做与运算
    int j = (hash >>> segmentShift) & segmentMask;
    if ((s = (Segment<K,V>)UNSAFE.getObject          // nonvolatile; recheck
         (segments, (j << SSHIFT) + SBASE)) == null) //  in ensureSegment
        // 如果查找到的 Segment 为空，初始化
        s = ensureSegment(j);
    return s.put(key, hash, value, false);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
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;
    // 判断 u 位置的 Segment 是否为null
    if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) {
        Segment<K,V> proto = ss[0]; // use segment 0 as prototype
        // 获取0号 segment 里的 HashEntry<K,V> 初始化长度
        int cap = proto.table.length;
        // 获取0号 segment 里的 hash 表里的扩容负载因子，所有的 segment 的 loadFactor 是相同的
        float lf = proto.loadFactor;
        // 计算扩容阀值
        int threshold = (int)(cap * lf);
        // 创建一个 cap 容量的 HashEntry 数组
        HashEntry<K,V>[] tab = (HashEntry<K,V>[])new HashEntry[cap];
        if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) { // recheck
            // 再次检查 u 位置的 Segment 是否为null，因为这时可能有其他线程进行了操作
            Segment<K,V> s = new Segment<K,V>(lf, threshold, tab);
            // 自旋检查 u 位置的 Segment 是否为null
            while ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
                   == null) {
                // 使用CAS 赋值，只会成功一次
                if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s))
                    break;
            }
        }
    }
    return seg;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
1. tryLock() 获取锁，获取不到使用 scanAndLockForPut 方法继续获取。
2. 计算 put 的数据要放入的 index 位置，然后获取这个位置上的 HashEntry 。
3. 遍历 put 新元素，为什么要遍历？因为这里获取的 HashEntry 可能是一个空元素，也可能是链表已存在，所以要区别对待。
3.1 如果这个位置上的 HashEntry 不存在：如果当前容量大于扩容阀值，小于最大容量，进行扩容。直接头插法插入。
3.2 如果这个位置上的 HashEntry 存在：判断链表当前元素 key 和 hash 值是否和要 put 的 key 和 hash 值一致。一致则替换值，不一致，获取链表下一个节点，直到发现相同进行值替换，或者链表表里完毕没有相同的。 如果当前容量大于扩容阀值，小于最大容量，进行扩容。直接链表头插法插入。
4. 如果要插入的位置之前已经存在，替换后返回旧值，否则返回 null.
final V put(K key, int hash, V value, boolean onlyIfAbsent) {
    // 获取 ReentrantLock 独占锁，获取不到，scanAndLockForPut 获取。
    HashEntry<K,V> node = tryLock() ? null : scanAndLockForPut(key, hash, value);
    V oldValue;
    try {
        HashEntry<K,V>[] tab = table;
        // 计算要put的数据位置
        int index = (tab.length - 1) & hash;
        // CAS 获取 index 坐标的值
        HashEntry<K,V> first = entryAt(tab, index);
        for (HashEntry<K,V> e = first;;) {
            if (e != null) {
                // 检查是否 key 已经存在，如果存在，则遍历链表寻找位置，找到后替换 value
                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 {
                // first 有值没说明 index 位置已经有值了，有冲突，链表头插法。
                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
                    // index 位置赋值 node，node 可能是一个元素，也可能是一个链表的表头
                    setEntryAt(tab, index, node);
                ++modCount;
                count = c;
                oldValue = null;
                break;
            }
        }
    } finally {
        unlock();
    }
    return oldValue;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
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;
        }
        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;
}

4. 扩容rehash()方法

原理
代码
流程图

扩容只会扩容到原来的两倍。老数组里的数据移动到新的数组时，位置要么不变，要么变为 index+ oldSize，参数里的 node 会在扩容之后使用链表头插法插入到指定位置。

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
private void rehash(HashEntry<K,V> node) {
    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];
    // 新的掩码，默认2扩容后是4，-1是3，二进制就是11。
    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;
                // 新的位置只可能是不便或者是老的位置+老的容量。
                // 遍历结束后，lastRun 后面的元素位置都是相同的
                for (HashEntry<K,V> 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<K,V> p = e; p != lastRun; p = p.next) {
                    // 遍历剩余元素，头插法到指定 k 位置。
                    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;
}

略

5. get()方法

原理
代码
  1. 计算得到 key 的存放位置
  2. 遍历指定位置查找相同 key 的 value 值
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
public V get(Object key) {
    Segment<K,V> s; // manually integrate access methods to reduce overhead
    HashEntry<K,V>[] tab;
    int h = hash(key);
    long u = (((h >>> segmentShift) & segmentMask) << SSHIFT) + SBASE;
    // 计算得到 key 的存放位置
    if ((s = (Segment<K,V>)UNSAFE.getObjectVolatile(segments, u)) != null &&
        (tab = s.table) != null) {
        for (HashEntry<K,V> e = (HashEntry<K,V>) UNSAFE.getObjectVolatile
                 (tab, ((long)(((tab.length - 1) & h)) << TSHIFT) + TBASE);
             e != null; e = e.next) {
            // 如果是链表，遍历查找到相同 key 的 value。
            K k;
            if ((k = e.key) == key || (e.hash == h && key.equals(k)))
                return e.value;
        }
    }
    return null;
}

2. ConcurrentHashMap1.8

1. 属性和方法

继承关系
常量
属性
内部类
unSafe()方法
继承关系
继承关系
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
// 哈希表的最大容量，采用位运算表示，1左移30位
private static final int MAXIMUM_CAPACITY = 1 << 30;

// 默认初始容量为16
private static final int DEFAULT_CAPACITY = 16;

// 数组的最大容量，Integer.MAX_VALUE减去8
static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;

// 默认并发级别为16
private static final int DEFAULT_CONCURRENCY_LEVEL = 16;

// 哈希表的负载因子，默认为0.75
private static final float LOAD_FACTOR = 0.75f;

// 树化阈值，即链表长度大于等于8时将链表转化为红黑树
static final int TREEIFY_THRESHOLD = 8;

// 退树化阈值，即红黑树节点个数小于等于6时将红黑树还原为链表
static final int UNTREEIFY_THRESHOLD = 6;

// 树化的最小容量，即哈希表的最小容量大于等于64时才考虑树化
static final int MIN_TREEIFY_CAPACITY = 64;

// 最小的转移步长，用于在扩容时确定扩容的步长
private static final int MIN_TRANSFER_STRIDE = 16;

// 用于标记哈希表的扩容状态的位数
private static int RESIZE_STAMP_BITS = 16;

// 最大的扩容线程数，即2^(32-RESIZE_STAMP_BITS) - 1
private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1;

// 用于计算扩容状态的位移
private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;

// 表示节点已经移动的状态标志，用于标记正在扩容的节点
static final int MOVED     = -1; // 用于标记转移节点的哈希值
static final int TREEBIN   = -2; // 用于标记树节点的哈希值
static final int RESERVED  = -3; // 用于标记临时保留的哈希值
static final int HASH_BITS = 0x7fffffff; // 正常节点哈希值的可用位数

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
// 用于存储键值对的哈希表，transient关键字表示不会被序列化
transient volatile Node<K,V>[] table;

// 用于在扩容时暂存新的哈希表
private transient volatile Node<K,V>[] nextTable;

// 基础计数器，用于记录实际存储的键值对数量
private transient volatile long baseCount;

// 哈希表的控制参数，用于控制哈希表的大小和扩容
private transient volatile int sizeCtl;

// 扩容时用于标记当前正在处理的槽的索引
private transient volatile int transferIndex;

// 用于标记当前正在处理的槽的索引，防止多线程同时进行扩容
private transient volatile int cellsBusy;

// 用于存储计数器的数组，用于处理高并发情况下的计数
private transient volatile CounterCell[] counterCells;

// 视图，用于提供对键集合、值集合、条目集合的视图
private transient KeySetView<K,V> keySet;
private transient ValuesView<K,V> values;
private transient EntrySetView<K,V> entrySet;
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
static class Node<K,V> implements Map.Entry<K,V> {
    final int hash;
    final K key;
    volatile V val;
    volatile Node<K,V> next;
}

static final class TreeNode<K,V> extends Node<K,V> {
    TreeNode<K,V> parent;  // red-black tree links
    TreeNode<K,V> left;
    TreeNode<K,V> right;
    TreeNode<K,V> prev;    // needed to unlink next upon deletion
    boolean red;
}

static class Segment<K,V> extends ReentrantLock implements Serializable {
    private static final long serialVersionUID = 2249069246763182397L;
    final float loadFactor;
    Segment(float lf) { this.loadFactor = lf; }
}

static final class TreeBin<K,V> extends Node<K,V> {
    TreeNode<K,V> root;
    volatile TreeNode<K,V> first;
    volatile Thread waiter;
    volatile int lockState;
    // values for lockState
    static final int WRITER = 1; // set while holding write lock
    static final int WAITER = 2; // set when waiting for write lock
    static final int READER = 4; // increment value for setting read lock
}

static final class ForwardingNode<K,V> extends Node<K,V> {
    final Node<K,V>[] nextTable;
    ForwardingNode(Node<K,V>[] tab) {
        super(MOVED, null, null, null);
        this.nextTable = tab;
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
/*
 * 用来返回节点数组的指定位置的节点的原子操作
 */
@SuppressWarnings("unchecked")
static final <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i) {
    return (Node<K,V>)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE);
}

/*
 * cas原子操作，在指定位置设定值
 */
static final <K,V> boolean casTabAt(Node<K,V>[] tab, int i,
                                    Node<K,V> c, Node<K,V> v) {
    return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v);
}
/*
 * 原子操作，在指定位置设定值
 */
static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) {
    U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v);
}

2. 存储结构

数组+链表+红黑树
数组+链表+红黑树

3. 初始化initTable方法

原理
代码

初始化数组table，如果sizeCtl小于0，说明别的线程正在进行初始化，则让出执行权,如果sizeCtl大于0的话，则初始化一个大小为sizeCtl的数组,否则的话初始化一个默认大小(16)的数组,然后设置sizeCtl的值为数组长度的3/4
sizeCtl说明：

  • -1 说明正在初始化，其他线程需要自旋等待
  • -N 说明 table 正在进行扩容，高 16 位表示扩容的标识戳，低 16 位减 1 为正在进行扩容的线程数
  • 如果table 没有初始化，表示 table 初始化大小
  • 如果table 已经初始化，表示 table 扩容的阈值，默认是容量的0.75倍
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
private final Node<K,V>[] initTable() {
    Node<K,V>[] tab; int sc;
    while ((tab = table) == null || tab.length == 0) {
        //　如果 sizeCtl < 0 ,说明另外的线程执行CAS 成功，正在进行初始化。
        if ((sc = sizeCtl) < 0)
            // 让出 CPU 使用权
            Thread.yield(); // lost initialization race; just spin
        else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
            try {
                if ((tab = table) == null || tab.length == 0) {
                    int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
                    @SuppressWarnings("unchecked")
                    Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                    table = tab = nt;
                    sc = n - (n >>> 2);
                }
            } finally {
                sizeCtl = sc;
            }
            break;
        }
    }
    return tab;
}

4. put()方法

原理
code-put()
code-putVal()
  1. 根据 key 计算出 hashcode 。
  2. 判断是否需要进行初始化。
  3. 当前 key 定位出的 Node，如果为空表示当前位置可以写入数据，利用 CAS 尝试写入，失败则自旋保证成功。
  4. 如果当前位置的 hashcode == MOVED == -1,则需要进行扩容。
  5. 如果都不满足，则利用 synchronized 锁写入数据。
  6. 如果数量大于 TREEIFY_THRESHOLD 则要执行树化方法，在 treeifyBin 中会首先判断当前数组长度 ≥64 时才会将链表转换为红黑树。
1
2
3
4
5
6
7
/**
* 当设置为false的时候表示这个value一定会设置
* true的时候，只有当这个key的value为空的时候才会设置
**/
public V put(K key, V value) {
    return putVal(key, value, false);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
/*
 * 当添加一对键值对的时候，首先会去判断保存这些键值对的数组是不是初始化了，
 * 如果没有的话就初始化数组
 *  然后通过计算hash值来确定放在数组的哪个位置
 * 如果这个位置为空则直接添加，如果不为空的话，则取出这个节点来
 * 如果取出来的节点的hash值是MOVED(-1)的话，则表示当前正在对这个数组进行扩容，复制到新的数组，则当前线程也去帮助复制
 * 最后一种情况就是，如果这个节点，不为空，也不在扩容，则通过synchronized来加锁，进行添加操作
 *    然后判断当前取出的节点位置存放的是链表还是树
 *    如果是链表的话，则遍历整个链表，直到取出来的节点的key来个要放的key进行比较，如果key相等，并且key的hash值也相等的话，
 *          则说明是同一个key，则覆盖掉value，否则的话则添加到链表的末尾
 *    如果是树的话，则调用putTreeVal方法把这个元素添加到树中去
 *  最后在添加完成之后，会判断在该节点处共有多少个节点（注意是添加前的个数），如果达到8个以上了的话，
 *  则调用treeifyBin方法来尝试将处的链表转为树，或者扩容数组
 */
final V putVal(K key, V value, boolean onlyIfAbsent) {
    if (key == null || value == null) throw new NullPointerException();//K,V都不能为空，否则的话跑出异常
    int hash = spread(key.hashCode());    //取得key的hash值
    int binCount = 0;    //用来计算在这个节点总共有多少个元素，用来控制扩容或者转移为树
    for (Node<K,V>[] tab = table;;) {    //
        Node<K,V> f; int n, i, fh;
        if (tab == null || (n = tab.length) == 0)    
            tab = initTable();    //第一次put的时候table没有初始化，则初始化table
        else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {    //通过哈希计算出一个表中的位置因为n是数组的长度，所以(n-1)&hash肯定不会出现数组越界
            if (casTabAt(tab, i, null,        //如果这个位置没有元素的话，则通过cas的方式尝试添加，注意这个时候是没有加锁的
                         new Node<K,V>(hash, key, value, null)))        //创建一个Node添加到数组中区，null表示的是下一个节点为空
                break;                   // no lock when adding to empty bin
        }
        /*
         * 如果检测到某个节点的hash值是MOVED，则表示正在进行数组扩张的数据复制阶段，
         * 则当前线程也会参与去复制，通过允许多线程复制的功能，一次来减少数组的复制所带来的性能损失
         */
        else if ((fh = f.hash) == MOVED)    
            tab = helpTransfer(tab, f);
        else {
            /*
             * 如果在这个位置有元素的话，就采用synchronized的方式加锁，
             *     如果是链表的话(hash大于0)，就对这个链表的所有元素进行遍历，
             *         如果找到了key和key的hash值都一样的节点，则把它的值替换到
             *         如果没找到的话，则添加在链表的最后面
             *  否则，是树的话，则调用putTreeVal方法添加到树中去
             *  
             *  在添加完之后，会对该节点上关联的的数目进行判断，
             *  如果在8个以上的话，则会调用treeifyBin方法，来尝试转化为树，或者是扩容
             */
            V oldVal = null;
            synchronized (f) {
                if (tabAt(tab, i) == f) {        //再次取出要存储的位置的元素，跟前面取出来的比较
                    if (fh >= 0) {                //取出来的元素的hash值大于0，当转换为树之后，hash值为-2
                        binCount = 1;            
                        for (Node<K,V> e = f;; ++binCount) {    //遍历这个链表
                            K ek;
                            if (e.hash == hash &&        //要存的元素的hash，key跟要存储的位置的节点的相同的时候，替换掉该节点的value即可
                                ((ek = e.key) == key ||
                                 (ek != null && key.equals(ek)))) {
                                oldVal = e.val;
                                if (!onlyIfAbsent)        //当使用putIfAbsent的时候，只有在这个key没有设置值得时候才设置
                                    e.val = value;
                                break;
                            }
                            Node<K,V> pred = e;
                            if ((e = e.next) == null) {    //如果不是同样的hash，同样的key的时候，则判断该节点的下一个节点是否为空，
                                pred.next = new Node<K,V>(hash, key,        //为空的话把这个要加入的节点设置为当前节点的下一个节点
                                                          value, null);
                                break;
                            }
                        }
                    }
                    else if (f instanceof TreeBin) {    //表示已经转化成红黑树类型了
                        Node<K,V> p;
                        binCount = 2;
                        if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,    //调用putTreeVal方法，将该元素添加到树中去
                                                       value)) != null) {
                            oldVal = p.val;
                            if (!onlyIfAbsent)
                                p.val = value;
                        }
                    }
                }
            }
            if (binCount != 0) {
                if (binCount >= TREEIFY_THRESHOLD)    //当在同一个节点的数目达到8个的时候，则扩张数组或将给节点的数据转为tree
                    treeifyBin(tab, i);    
                if (oldVal != null)
                    return oldVal;
                break;
            }
        }
    }
    addCount(1L, binCount);    //计数
    return null;
}

5. get()方法

原理
代码
  1. 根据 hash 值计算位置。
  2. 查找到指定位置，如果头节点就是要找的，直接返回它的 value.
  3. 如果头节点 hash 值小于 0 ，说明正在扩容或者是红黑树，查找之。
  4. 如果是链表，遍历查找之。
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
public V get(Object key) {
    Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
    // key 所在的 hash 位置
    int h = spread(key.hashCode());
    if ((tab = table) != null && (n = tab.length) > 0 &&
        (e = tabAt(tab, (n - 1) & h)) != null) {
        // 如果指定位置元素存在，头结点hash值相同
        if ((eh = e.hash) == h) {
            if ((ek = e.key) == key || (ek != null && key.equals(ek)))
                // key hash 值相等，key值相同，直接返回元素 value
                return e.val;
        }
        else if (eh < 0)
            // 头结点hash值小于0，说明正在扩容或者是红黑树，find查找
            return (p = e.find(h, key)) != null ? p.val : null;
        while ((e = e.next) != null) {
            // 是链表，遍历查找
            if (e.hash == h &&
                ((ek = e.key) == key || (ek != null && key.equals(ek))))
                return e.val;
        }
    }
    return null;
}

6. 扩容rehash()方法

原理
code-treeifyBin()-扩容 or 变树
code-tryPresize()-扩容
code-transfer()-扩容核心逻辑

注意：

  1. 复制之后的新链表不是旧链表的绝对倒序。
  2. 在扩容的时候每个线程都有处理的步长，最少为16，在这个步长范围内的数组节点只有自己一个线程来处理
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
/**
 * Replaces all linked nodes in bin at given index unless table is
 * too small, in which case resizes instead.
 * 当数组长度小于64的时候，扩张数组长度一倍，否则的话把链表转为树
 */
private final void treeifyBin(Node<K,V>[] tab, int index) {
    Node<K,V> b; int n, sc;
    if (tab != null) {
            System.out.println("treeifyBin方\t==>数组长："+tab.length);
        if ((n = tab.length) < MIN_TREEIFY_CAPACITY)    //MIN_TREEIFY_CAPACITY 64
            tryPresize(n << 1);        // 数组扩容
        else if ((b = tabAt(tab, index)) != null && b.hash >= 0) {
            synchronized (b) {    //使用synchronized同步器，将该节点出的链表转为树
                if (tabAt(tab, index) == b) {
                    TreeNode<K,V> hd = null, tl = null;    //hd：树的头(head)
                    for (Node<K,V> e = b; e != null; e = e.next) {
                        TreeNode<K,V> p =
                            new TreeNode<K,V>(e.hash, e.key, e.val,
                                              null, null);
                        if ((p.prev = tl) == null)        //把Node组成的链表，转化为TreeNode的链表，头结点任然放在相同的位置
                            hd = p;    //设置head
                        else
                            tl.next = p;
                        tl = p;
                    }
                    setTabAt(tab, index, new TreeBin<K,V>(hd));//把TreeNode的链表放入容器TreeBin中
                }
            }
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
/**
 * 扩容表为指可以容纳指定个数的大小（总是2的N次方）
 * 假设原来的数组长度为16，则在调用tryPresize的时候，size参数的值为16<<1(32)，此时sizeCtl的值为12
 * 计算出来c的值为64,则要扩容到sizeCtl≥为止
 *  第一次扩容之后 数组长：32 sizeCtl：24
 *  第二次扩容之后 数组长：64 sizeCtl：48
 *  第二次扩容之后 数组长：128 sizeCtl：94 --> 这个时候才会退出扩容
 */
private final void tryPresize(int size) {
        /*
         * MAXIMUM_CAPACITY = 1 << 30
         * 如果给定的大小大于等于数组容量的一半，则直接使用最大容量，
         * 否则使用tableSizeFor算出来
         * 后面table一直要扩容到这个值小于等于sizeCtrl(数组长度的3/4)才退出扩容
         */
    int c = (size >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY :
        tableSizeFor(size + (size >>> 1) + 1);
    int sc;
    while ((sc = sizeCtl) >= 0) {
        Node<K,V>[] tab = table; int n;
          printTable(tab);    调试用的
        /*
         * 如果数组table还没有被初始化，则初始化一个大小为sizeCtrl和刚刚算出来的c中较大的一个大小的数组
         * 初始化的时候，设置sizeCtrl为-1，初始化完成之后把sizeCtrl设置为数组长度的3/4
         * 为什么要在扩张的地方来初始化数组呢？这是因为如果第一次put的时候不是put单个元素，
         * 而是调用putAll方法直接put一个map的话，在putALl方法中没有调用initTable方法去初始化table，
         * 而是直接调用了tryPresize方法，所以这里需要做一个是不是需要初始化table的判断
         */
        if (tab == null || (n = tab.length) == 0) {
            n = (sc > c) ? sc : c;
            if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {    //初始化tab的时候，把sizeCtl设为-1
                try {
                    if (table == tab) {
                        @SuppressWarnings("unchecked")
                        Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                        table = nt;
                        sc = n - (n >>> 2);
                    }
                } finally {
                    sizeCtl = sc;
                }
            }
        }
        /*
         * 一直扩容到的c小于等于sizeCtl或者数组长度大于最大长度的时候，则退出
         * 所以在一次扩容之后，不是原来长度的两倍，而是2的n次方倍
         */
        else if (c <= sc || n >= MAXIMUM_CAPACITY) {
                break;    //退出扩张
        }
        else if (tab == table) {
            int rs = resizeStamp(n);
            /*
             * 如果正在扩容Table的话，则帮助扩容
             * 否则的话，开始新的扩容
             * 在transfer操作，将第一个参数的table中的元素，移动到第二个元素的table中去，
             * 虽然此时第二个参数设置的是null，但是，在transfer方法中，当第二个参数为null的时候，
             * 会创建一个两倍大小的table
             */
            if (sc < 0) {
                Node<K,V>[] nt;
                if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
                    sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
                    transferIndex <= 0)
                    break;
                /*
                 * transfer的线程数加一,该线程将进行transfer的帮忙
                 * 在transfer的时候，sc表示在transfer工作的线程数
                 */
                if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
                    transfer(tab, nt);
            }
            /*
             * 没有在初始化或扩容，则开始扩容
             */
            else if (U.compareAndSwapInt(this, SIZECTL, sc,
                                         (rs << RESIZE_STAMP_SHIFT) + 2)) {
                    transfer(tab, null);
            }
        }
    }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
/**
 * Moves and/or copies the nodes in each bin to new table. See
 * above for explanation.
 * 把数组中的节点复制到新的数组的相同位置，或者移动到扩张部分的相同位置
 * 在这里首先会计算一个步长，表示一个线程处理的数组长度，用来控制对CPU的使用，
 * 每个CPU最少处理16个长度的数组元素,也就是说，如果一个数组的长度只有16，那只有一个线程会对其进行扩容的复制移动操作
 * 扩容的时候会一直遍历，知道复制完所有节点，没处理一个节点的时候会在链表的头部设置一个fwd节点，这样其他线程就会跳过他，
 * 复制后在新数组中的链表不是绝对的反序的
 */
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
    int n = tab.length, stride;
    if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)    //MIN_TRANSFER_STRIDE 用来控制不要占用太多CPU
        stride = MIN_TRANSFER_STRIDE; // subdivide range    //MIN_TRANSFER_STRIDE=16
    /*
     * 如果复制的目标nextTab为null的话，则初始化一个table两倍长的nextTab
     * 此时nextTable被设置值了(在初始情况下是为null的)
     * 因为如果有一个线程开始了表的扩张的时候，其他线程也会进来帮忙扩张，
     * 而只是第一个开始扩张的线程需要初始化下目标数组
     */
    if (nextTab == null) {            // initiating
        try {
            @SuppressWarnings("unchecked")
            Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
            nextTab = nt;
        } catch (Throwable ex) {      // try to cope with OOME
            sizeCtl = Integer.MAX_VALUE;
            return;
        }
        nextTable = nextTab;
        transferIndex = n;
    }
    int nextn = nextTab.length;
    /*
     * 创建一个fwd节点，这个是用来控制并发的，当一个节点为空或已经被转移之后，就设置为fwd节点
     * 这是一个空的标志节点
     */
    ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
    boolean advance = true;    //是否继续向前查找的标志位
    boolean finishing = false; // to ensure sweep(清扫) before committing nextTab,在完成之前重新在扫描一遍数组，看看有没完成的没
    for (int i = 0, bound = 0;;) {
        Node<K,V> f; int fh;
        while (advance) {
            int nextIndex, nextBound;
            if (--i >= bound || finishing) {
                advance = false;
            }
            else if ((nextIndex = transferIndex) <= 0) {
                i = -1;
                advance = false;
            }
            else if (U.compareAndSwapInt
                     (this, TRANSFERINDEX, nextIndex,
                      nextBound = (nextIndex > stride ?
                                   nextIndex - stride : 0))) {
                bound = nextBound;
                i = nextIndex - 1;
                advance = false;
            }
        }
        if (i < 0 || i >= n || i + n >= nextn) {
            int sc;
            if (finishing) {        //已经完成转移
                nextTable = null;
                table = nextTab;
                sizeCtl = (n << 1) - (n >>> 1);    //设置sizeCtl为扩容后的0.75
                return;
            }
            if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
                if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT) {
                        return;
                }
                finishing = advance = true;
                i = n; // recheck before commit
            }
        }
        else if ((f = tabAt(tab, i)) == null)            //数组中把null的元素设置为ForwardingNode节点(hash值为MOVED[-1])
            advance = casTabAt(tab, i, null, fwd);
        else if ((fh = f.hash) == MOVED)
            advance = true; // already processed
        else {
            synchronized (f) {                //加锁操作
                if (tabAt(tab, i) == f) {
                    Node<K,V> ln, hn;
                    if (fh >= 0) {        //该节点的hash值大于等于0，说明是一个Node节点
                            /*
                             * 因为n的值为数组的长度，且是power(2,x)的，所以，在&操作的结果只可能是0或者n
                             * 根据这个规则
                             *         0-->  放在新表的相同位置
                             *         n-->  放在新表的（n+原来位置）
                             */
                        int runBit = fh & n; 
                        Node<K,V> lastRun = f;
                        /*
                         * lastRun 表示的是需要复制的最后一个节点
                         * 每当新节点的hash&n -> b 发生变化的时候，就把runBit设置为这个结果b
                         * 这样for循环之后，runBit的值就是最后不变的hash&n的值
                         * 而lastRun的值就是最后一次导致hash&n 发生变化的节点(假设为p节点)
                         * 为什么要这么做呢？因为p节点后面的节点的hash&n 值跟p节点是一样的，
                         * 所以在复制到新的table的时候，它肯定还是跟p节点在同一个位置
                         * 在复制完p节点之后，p节点的next节点还是指向它原来的节点，就不需要进行复制了，自己就被带过去了
                         * 这也就导致了一个问题就是复制后的链表的顺序并不一定是原来的倒序
                         */
                        for (Node<K,V> p = f.next; p != null; p = p.next) {
                            int b = p.hash & n;    //n的值为扩张前的数组的长度
                            if (b != runBit) {
                                runBit = b;
                                lastRun = p;
                            }
                        }
                        if (runBit == 0) {
                            ln = lastRun;
                            hn = null;
                        }
                        else {
                            hn = lastRun;
                            ln = null;
                        }
                        /*
                         * 构造两个链表，顺序大部分和原来是反的
                         * 分别放到原来的位置和新增加的长度的相同位置(i/n+i)
                         */
                        for (Node<K,V> p = f; p != lastRun; p = p.next) {
                            int ph = p.hash; K pk = p.key; V pv = p.val;
                            if ((ph & n) == 0)
                                    /*
                                     * 假设runBit的值为0，
                                     * 则第一次进入这个设置的时候相当于把旧的序列的最后一次发生hash变化的节点(该节点后面可能还有hash计算后同为0的节点)设置到旧的table的第一个hash计算后为0的节点下一个节点
                                     * 并且把自己返回，然后在下次进来的时候把它自己设置为后面节点的下一个节点
                                     */
                                ln = new Node<K,V>(ph, pk, pv, ln);
                            else
                                    /*
                                     * 假设runBit的值不为0，
                                     * 则第一次进入这个设置的时候相当于把旧的序列的最后一次发生hash变化的节点(该节点后面可能还有hash计算后同不为0的节点)设置到旧的table的第一个hash计算后不为0的节点下一个节点
                                     * 并且把自己返回，然后在下次进来的时候把它自己设置为后面节点的下一个节点
                                     */
                                hn = new Node<K,V>(ph, pk, pv, hn);    
                        }
                        setTabAt(nextTab, i, ln);    
                        setTabAt(nextTab, i + n, hn);
                        setTabAt(tab, i, fwd);
                        advance = true;
                    }
                    else if (f instanceof TreeBin) {    //否则的话是一个树节点
                        TreeBin<K,V> t = (TreeBin<K,V>)f;
                        TreeNode<K,V> lo = null, loTail = null;
                        TreeNode<K,V> hi = null, hiTail = null;
                        int lc = 0, hc = 0;
                        for (Node<K,V> e = t.first; e != null; e = e.next) {
                            int h = e.hash;
                            TreeNode<K,V> p = new TreeNode<K,V>
                                (h, e.key, e.val, null, null);
                            if ((h & n) == 0) {
                                if ((p.prev = loTail) == null)
                                    lo = p;
                                else
                                    loTail.next = p;
                                loTail = p;
                                ++lc;
                            }
                            else {
                                if ((p.prev = hiTail) == null)
                                    hi = p;
                                else
                                    hiTail.next = p;
                                hiTail = p;
                                ++hc;
                            }
                        }
                        /*
                         * 在复制完树节点之后，判断该节点处构成的树还有几个节点，
                         * 如果≤6个的话，就转回为一个链表
                         */
                        ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) :
                            (hc != 0) ? new TreeBin<K,V>(lo) : t;
                        hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) :
                            (lc != 0) ? new TreeBin<K,V>(hi) : t;
                        setTabAt(nextTab, i, ln);
                        setTabAt(nextTab, i + n, hn);
                        setTabAt(tab, i, fwd);
                        advance = true;
                    }
                }
            }
        }
    }
}

7. concurrentHashMap的同步机制

7.1 什么情况下会引起数组的扩容？

扩容是通过transfer方法来进行的。而调用transfer方法的只有tryPresize、helpTransfer和addCount三个方法：

  1. tryPresize是在treeIfybin和putAll方法中调用，treeIfybin主要是在put添加元素完之后，判断该数组节点相关元素是不是已经超过8个的时候，如果超过则会调用这个方法来扩容数组或者把链表转为树。
  2. helpTransfer是在当一个线程要对table中元素进行操作的时候，如果检测到节点的HASH值为MOVED的时候，就会调用helpTransfer方法，在helpTransfer中再调用transfer方法来帮助完成数组的扩容.
  3. addCount是在当对数组进行操作，使得数组中存储的元素个数发生了变化的时候会调用的方法。

所以引起数组扩容的情况如下：

  1. 只有在往map中添加元素的时候，在某一个节点的数目已经超过了8个，同时数组的长度又小于64的时候，才会触发数组的扩容。
  2. 当数组中元素达到了sizeCtl的数量的时候，则会调用transfer方法来进行扩容
7.2 ConcurrentHashMap是如何做到并发安全，又是如何做到高效的并发的呢？

  • 读操作，从源码中可以看出来，在get操作中，根本没有使用同步机制，也没有使用unsafe方法，所以读操作是支持并发操作的。

  • 写操作，同步处理主要是通过Synchronized和unsafe两种方式来完成的。

    • 在取得sizeCtl、某个位置的Node的时候，使用的都是unsafe的方法，来达到并发安全的目的
    • 当需要在某个位置设置节点的时候，则会通过Synchronized的同步机制来锁定该位置的节点。
    • 在数组扩容的时候，则通过处理的步长和fwd节点来达到并发安全的目的，通过设置hash值为MOVED
    • 当把某个位置的节点复制到扩张后的table的时候，也通过Synchronized的同步机制来保证现程安全

  • 在扩容的时候，可以不可以对数组进行读写操作呢？

    事实上是可以的。当在进行数组扩容的时候，如果当前节点还没有被处理（也就是说还没有设置为fwd节点），那就可以进行设置操作。如果该节点已经被处理了，则当前线程也会加入到扩容的操作中去。

3. concurrentHashMap在jdk1.7和jdk1.8中的差异点

3.1 数据结构

  • jdk7:分段锁
  • jdk8:数组+链表+红黑树

3.2 并发度

  • jdk7:16
  • jdk8:理想情况下是 table数组元素的个数

3.3 保证并发安全的原理

  • jdk7:segment分段锁，Segment 是继承自 ReentrantLock
  • jdk8:Node + CAS + synchronized

3.4 查询时间复杂度

  • jdk7:O(n)
  • jdk8:O(log(n))

3.5 hash碰撞的处理方式

  • jdk7:拉链法
  • jdk8:先使用拉链法，在链表长度超过一定阈值时，将链表转换为红黑树，来提高查找效率
参考资料

快来参与讨论吧~

博客近期分类标签归档
项目软件开发人工智能Staller
社交友链
更多关于本站
本站由 卡卡龙 使用 Stellar 1.29.1主题创建

本站访问量 次. 本文阅读量 次.

Stellar Copyright Licenses Hexo 备案 描述文字 描述文字