JDK1.8下ConcurrentHashMap源码解析

发表于

JDK1.8下的ConcurrentHashMap源码解析

上一篇博客介绍了HashMap的源码构成，现在我们趁热打铁继续看看ConcurrentHashMap吧。

简介

ConcurrentHashMap与HashMap的区别，就是ConcurrentHashMap是线程安全的。简单提一下HashTable，虽然HashTable也是线程安全的，HashTbale实现线程安全是通过用关键字synchronized修饰每个方法来实现的，当线程访问的方法正在被其他线程访问时，此线程必须进入阻塞或轮询状态，这就必然导致效率低下。 如线程1使用put进行添加元素，线程2不但不能使用put方法添加元素，并且也不能使用get方法来获取元素，所以竞争越激烈效率越低。下面就来介绍ConcurrentHashMap的优势之处。

ConcurrentHashMap在JAVA7中的实现原理

java7中ConcurrentHashMap结构图

java8之前，ConcurrentHashMap是将数据分成一段一段存储的，给每一段数据配一把锁，当一个线程获得锁互斥访问一段数据时，其他段的数据也可以被其他线程访问。ConcurrentHashMap结构：每一个segment（分段锁）都是一个HashEntry<K,V>[] table， table中的每一个元素本质上都是一个HashEntry的单向队列（单向链表实现）。

ConcurrentHashMap在JAVA8中的实现原理

java8中ConcurrentHashMap结构图.png

由图可见，Java8中实现已经抛弃了Segment分段锁机制，降低了锁的粒度，从对整个segment上锁到对Node上锁。利用CAS+Synchronized来保证并发更新的安全，底层依然采用数组+链表+红黑树的存储结构。

变量属性

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
// 最大容量
private static final int MAXIMUM_CAPACITY = 1 << 30;

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

// 最大数组大小 MAX_VALUE=2^31-1=2147483647
static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;

// 默认并发数
private static final int DEFAULT_CONCURRENCY_LEVEL = 16;

// 负载因子
private static final float LOAD_FACTOR = 0.75f;

// 链表转树的阀值，如果table[i]下面的链表长度大于8时就转化为树
static final int TREEIFY_THRESHOLD = 8;

//树转链表的阀值，小于等于6是转为链表，仅在扩容tranfer时才可能树转链表
static final int UNTREEIFY_THRESHOLD = 6;

// 转为红黑树table的最小容量，小于此值就resize
static final int MIN_TREEIFY_CAPACITY = 64;

// 每次进行转移的最小值
private static final int MIN_TRANSFER_STRIDE = 16;

// 生成sizeCtl所使用的bit位数
private static int RESIZE_STAMP_BITS = 16;

// help resize的最大线程数
private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1;

// 记录sizeCtl中的大小所需要进行的偏移位数
private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;

// 标识
static final int MOVED     = -1; // hash for forwarding nodes（forwarding nodes的hash值）、标示位
static final int TREEBIN   = -2; // hash for roots of trees（树根节点的hash值）
static final int RESERVED  = -3; // hash for transient reservations（ReservationNode的hash值）
static final int HASH_BITS = 0x7fffffff; // usable bits of normal node hash

// 获取可用的CPU个数
static final int NCPU = Runtime.getRuntime().availableProcessors();

// 进行序列化的属性
private static final ObjectStreamField[] serialPersistentFields = {
    new ObjectStreamField("segments", Segment[].class),
    new ObjectStreamField("segmentMask", Integer.TYPE),
    new ObjectStreamField("segmentShift", Integer.TYPE)
};
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 volatile Node<K,V>[] table;

// rehash扩容时用到的新键值对数组
private transient volatile Node<K,V>[] nextTable;

// 记录当前键值对总数，通过CAS更新，对所有线程可见
private transient volatile long baseCount;

// 对表初始化和扩容控制
private transient volatile int sizeCtl;

// 扩容下另一个表的索引
private transient volatile int transferIndex;

// 自旋锁
private transient volatile int cellsBusy;

// counter cell表，长度总为2的幂次
private transient volatile CounterCell[] counterCells;

// 视图
private transient KeySetView<K,V> keySet;
private transient ValuesView<K,V> values;
private transient EntrySetView<K,V> entrySet;

这里重点提一下sizeCtl变量，sizeCtl表示键值对总数阈值，通过CAS更新, 对所有线程可见

当sizeCtl < 0时，表示多个线程在等待扩容；

当sizeCtl = 0时，默认值；

当sizeCtl > 0时，表示扩容的阈值；

重要方法解析

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
static class Node<K,V> implements Map.Entry<K,V> {
    final int hash;
    final K key;
    // java8中这两个参数增加volatile修饰，保证数据的可视性
    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;
    }

    public final K getKey()       { return key; }
    public final V getValue()     { return val; }
    public final int hashCode()   { return key.hashCode() ^ val.hashCode(); }
    public final String toString(){ return key + "=" + val; }
    // 不允许修改value的值，在hashMap中是允许的
    public final V setValue(V value) {
        throw new UnsupportedOperationException();
    }

    public final boolean equals(Object o) {
        Object k, v, u; Map.Entry<?,?> e;
        return ((o instanceof Map.Entry) &&
                (k = (e = (Map.Entry<?,?>)o).getKey()) != null &&
                (v = e.getValue()) != null &&
                (k == key || k.equals(key)) &&
                (v == (u = val) || v.equals(u)));
    }

    /**
     * Virtualized support for map.get(); overridden in subclasses.
     */
    Node<K,V> find(int h, Object k) {
        Node<K,V> e = this;
        if (k != null) {
            do {
                K ek;
                if (e.hash == h &&
                    ((ek = e.key) == k || (ek != null && k.equals(ek))))
                    return e;
            } while ((e = e.next) != null);
        }
        return null;
    }
}

这个Node内部类和HashMap的Node内部类很相似，但还是有以下区别

1：此Node类中对value和next属性设置了volatile同步锁

2：此Node类中不允许setValue方法直接改变Node的value域

3：此Node类中增加了find方法辅助map.get()方法

volatile关键字
这里介绍一下volatile关键词。volatile关键词可确保应用中的可视性，如果你将一个域声明为volatile的，那么只要对这个域产生了写操作，那么所有的读操作就能看见这个修改。即便使用了本地缓存，情况也是如此，volatile域会立即被写入到主存中，而读取操作就发生到主存中。

理解原子性和易变形是不同的概念这一点很重要。而非volatile域上的原子操作不必刷新到主存中去，因此其他读取该域的任务也不必看到这个新值。如果多个任务在同时访问某个域，那么这个域就应该时volatile的，否则，这个域就应该只能经由同步来访问。同步也会导致向主存中刷新，因此如果一个域完全由synchronized方法或语句块来防护，那就不必将其设置为是volatile的。
TreeNode() —— Node for use in TreeBins
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
//  HashMap的TreeNode继承至LinkedHashMap.Entry；而这里继承至自己实现的Node，将带有next指针，便于treebin访问。
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;

    TreeNode(int hash, K key, V val, Node<K,V> next,
             TreeNode<K,V> parent) {
        super(hash, key, val, next);
        this.parent = parent;
    }

    Node<K,V> find(int h, Object k) {
        return findTreeNode(h, k, null);
    }

    // 查找hash为h，key为k的节点
    final TreeNode<K,V> findTreeNode(int h, Object k, Class<?> kc) {
        if (k != null) {// hashMap这里没有判空
            TreeNode<K,V> p = this;
            do  {
                int ph, dir; K pk; TreeNode<K,V> q;
                TreeNode<K,V> pl = p.left, pr = p.right;
                if ((ph = p.hash) > h)
                    p = pl;
                else if (ph < h)
                    p = pr;
                else if ((pk = p.key) == k || (pk != null && k.equals(pk)))
                    return p;
                else if (pl == null)
                    p = pr;
                else if (pr == null)
                    p = pl;
                else if ((kc != null ||
                          (kc = comparableClassFor(k)) != null) &&
                         (dir = compareComparables(kc, k, pk)) != 0)
                    p = (dir < 0) ? pl : pr;
                else if ((q = pr.findTreeNode(h, k, kc)) != null)
                    return q;
                else
                    p = pl;
            } while (p != null);
        }
        return null;
    }
}
与HashMap相比，这里的TreeNode相当简洁；ConcurrentHashMap链表转树时，并不会直接转，正如注释（Nodes for use in TreeBins）所说，只是把这些节点包装成TreeNode放到TreeBin中，再由TreeBin来转化红黑树。
TreeBin() —— 封装TreeNode
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
// TreeBin用于封装维护TreeNode，包含putTreeVal、lookRoot、UNlookRoot、remove、balanceInsetion、balanceDeletion等方法，这里只分析其构造函数。  
// 当链表转树时，用于封装TreeNode，也就是说，ConcurrentHashMap的红黑树存放的时TreeBin，而不是treeNode。  
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

    /**
     * Tie-breaking utility for ordering insertions when equal
     * hashCodes and non-comparable. We don't require a total
     * order, just a consistent insertion rule to maintain
     * equivalence across rebalancings. Tie-breaking further than
     * necessary simplifies testing a bit.
     */
    static int tieBreakOrder(Object a, Object b) {
        int d;
        if (a == null || b == null ||
            (d = a.getClass().getName().
             compareTo(b.getClass().getName())) == 0)
            d = (System.identityHashCode(a) <= System.identityHashCode(b) ?
                 -1 : 1);
        return d;
    }

    /**
     * Creates bin with initial set of nodes headed by b.
     */
    TreeBin(TreeNode<K,V> b) {
        super(TREEBIN, null, null, null);
        this.first = b;
        TreeNode<K,V> r = null;
        for (TreeNode<K,V> x = b, next; x != null; x = next) {
            next = (TreeNode<K,V>)x.next;
            x.left = x.right = null;
            if (r == null) {
                x.parent = null;
                x.red = false;
                r = x;
            }
            else {
                K k = x.key;
                int h = x.hash;
                Class<?> kc = null;
                for (TreeNode<K,V> p = r;;) {
                    int dir, ph;
                    K pk = p.key;
                    if ((ph = p.hash) > h)
                        dir = -1;
                    else if (ph < h)
                        dir = 1;
                    else if ((kc == null &&
                              (kc = comparableClassFor(k)) == null) ||
                             (dir = compareComparables(kc, k, pk)) == 0)
                        dir = tieBreakOrder(k, pk);
                        TreeNode<K,V> xp = p;
                    if ((p = (dir <= 0) ? p.left : p.right) == null) {
                        x.parent = xp;
                        if (dir <= 0)
                            xp.left = x;
                        else
                            xp.right = x;
                        r = balanceInsertion(r, x);
                        break;
                    }
                }
            }
        }
        this.root = r;
        assert checkInvariants(root);
    }
putVal()
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
final V putVal(K key, V value, boolean onlyIfAbsent) {
    if (key == null || value == null) throw new NullPointerException();
    int hash = spread(key.hashCode());// 计算hash值
    int binCount = 0;
    // 不断CAS探测，如果其他线程正在修改tab，CAS尝试失败，知道成功为止
    // 这是一个无限循环，就是不断的尝试，因为在table的初始化和casTabAt用到了compareAndSwapInt、compareAndSwapObject
	// 因为table是用volatile修饰，如果其他线程正在修改tab，那么尝试就会失败，所以这边要加一个for循环，不断的尝试
    for (Node<K,V>[] tab = table;;) {
        Node<K,V> f; int n, i, fh;
        //空表，即对tab进行初始化
        if (tab == null || (n = tab.length) == 0)
            tab = initTable();
        // CAS探测空桶：表不为空且表的长度大于0，且该桶不为空，i即为节点位置
        else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
        // 比较并且交换值，如tab的第i项为空则用新生成的node替换
            if (casTabAt(tab, i, null,
                         new Node<K,V>(hash, key, value, null)))
                break;                   // no lock when adding to empty bin
        }
        // 当节点的hash值为MOVED时，即检测到tab[i]桶正在进行rehash
        else if ((fh = f.hash) == MOVED)
        // 进行节点转移（在扩容过程中）
            tab = helpTransfer(tab, f);
        else {
            V oldVal = null;
            //这个地方设计非常的巧妙，内置锁synchronized锁住了f,因为f是指定特定的tab[i]的，
            // 所以就锁住了整行链表,这个设计跟分段锁有异曲同工之妙，只是其他读取操作需要用cas来保证
            synchronized (f) {
                if (tabAt(tab, i) == f) {// 找到table表下标为i的节点
                    if (fh >= 0) {// table表中该节点的hash值大于0，即桶中键值对组织形式为链表
                        binCount = 1;
                        for (Node<K,V> e = f;; ++binCount) {// 无限循环
                            K ek;
                            if (e.hash == hash &&
                                ((ek = e.key) == key ||
                                // 节点hash值相等并且key也相等
                                 (ek != null && key.equals(ek)))) {
                                 // 保存该节点的val值
                                oldVal = e.val;
                                if (!onlyIfAbsent)// 查找到对应键值对
                                	// 将指定的value保存至节点。即进行节点值的更新
                                    e.val = value;
                                break;
                            }
                            // 保存当前节点
                            Node<K,V> pred = e;
                            // 当前节点的下一个节点为空，即为最后一个节点
                            if ((e = e.next) == null) {
                            	// 新生一个节点并且赋值给next域
                                pred.next = new Node<K,V>(hash, key,
                                                          value, null);
                                break;
                            }
                        }
                    }
                    else if (f instanceof TreeBin) {// 节点为红黑树类型
                        Node<K,V> p;
                        binCount = 2;
                        // 将hash、key、value放入红黑树
                        if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
                                                       value)) != null) {
                            // 保存节点的val
                            oldVal = p.val;
                            if (!onlyIfAbsent)
                                p.val = value;
                        }
                    }
                }
            }
            if (binCount != 0) {
                if (binCount >= TREEIFY_THRESHOLD)// 如果binCount大于等于转化为红黑树的阈值
                    treeifyBin(tab, i);// 转化为红黑树
                if (oldVal != null)
                    return oldVal;
                break;
            }
        }
    }
    //增加binCount的数量
    addCount(1L, binCount);
    return null;
}

putVal()方法中，最重要地方就是synchronized(f){}操作通过对桶的首元素 = 链表表头 Or 红黑树根节点加锁，从而实现对整个桶进行加锁，有锁分离思想的体现。

table中的三个原子操作
1
2
3
4
5
6
7
8
9
10
11
12
static final <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i) {// 获取索引i处Node
    return (Node<K,V>)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE);
}
// 利用CAS算法设置i位置上的Node节点（将c和table[i]比较，相同则插入v）
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);
}
// 仅在synchronized同步快中被调用，更新键值对
static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) {
    U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v);
}
addCount() —— CAS更新baseCount
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
private final void addCount(long x, int check) {
    CounterCell[] as; long b, s;
    // s = b + x,完成baseCount++操作
    if ((as = counterCells) != null ||
        !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
        CounterCell a; long v; int m;
        boolean uncontended = true;
        if (as == null || (m = as.length - 1) < 0 ||
            (a = as[ThreadLocalRandom.getProbe() & m]) == null ||
            !(uncontended =
              U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
              // 多线程CAS发生失败时执行
            fullAddCount(x, uncontended);
            return;
        }
        if (check <= 1)
            return;
        s = sumCount();
    }
    if (check >= 0) {
        Node<K,V>[] tab, nt; int n, sc;
        // 当更新后的键值对总数baseCount >= 阈值sizeCtl时， 进行rehash
        while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
               (n = tab.length) < MAXIMUM_CAPACITY) {
            int rs = resizeStamp(n);
            // sc < 0 表示其他线程已经在rehash
            if (sc < 0) {
                if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
                    sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
                    transferIndex <= 0)
                    break;
                // 其他线程的rehash操作已经完成，当前线程可以进行rehash
                if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
                    transfer(tab, nt);
            }
            // sc >= 0 表示只有当前线程在进行rehash操作，调用辅助扩容方法transfer
            else if (U.compareAndSwapInt(this, SIZECTL, sc,
                                         (rs << RESIZE_STAMP_SHIFT) + 2))
                transfer(tab, null);
            s = sumCount();
        }
    }
}
get()
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
public V get(Object key) {
    Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
    int h = spread(key.hashCode());
    if ((tab = table) != null && (n = tab.length) > 0 &&
        (e = tabAt(tab, (n - 1) & h)) != null) {
        if ((eh = e.hash) == h) {// 当表中元素的hash值与key的hash值相等时
            if ((ek = e.key) == key || (ek != null && key.equals(ek)))//键相等
                return e.val;
        }
        else if (eh < 0)// 节点hash值小于0
        	// 在桶（链表/红黑树）中查找
            return (p = e.find(h, key)) != null ? p.val : null;
        while ((e = e.next) != null) {// 对于节点hash值大于0的情况
            if (e.hash == h &&
                ((ek = e.key) == key || (ek != null && key.equals(ek))))
                return e.val;
        }
    }
    return null;
}

get函数根据的hash值来计算在哪个桶中，再遍历桶，查找元素，若找到则返回该结点，否则，返回null。