ConcurrentHashMap

Page content

简介

HashTable

HashMap 在多线程情况下,put 的时候,插入的元素超过了容量(由负载因子决定)的范围就会触发扩容操作,就是 rehash,这个会重新将原数组的内容重新 hash 到新的扩容数组中,在多线程的环境下,存在同时其他的元素也在进行 put 操作,如果 hash 值相同,可能出现同时在同一数组下用链表表示,造成闭环(循环链表),导致在 get 时会出现死循环,所以 HashMap 是线程不安全的。

HashTable 也是个键值存储集合,它是线程安全的,它在所有涉及到多线程操作上都加上了synchronized关键字来锁住整个table,这就意味着所有的线程都在竞争一把锁,在多线程的环境下,它是安全的,但是无疑是效率低下的。

其实 HashTable 有很大的优化空间,锁住整个 table 这么粗暴的方式可以变相的柔和点,比如在多线程的环境下,对不同的数据集进行操作时其实根本就不需要去竞争一个锁,因为它们 hash 值不同,不会因为rehash造成线程不安全,所以互不影响,这就是锁分离技术,将锁的粒度降低,利用多个锁来控制多个小的 table。

ConcurrentHashMap JDK1.8的实现

JDK1.8 的实现已经摒弃了 Segment 的概念,而是直接用Node数组+链表+红黑树的数据结构来实现,并发控制使用 SynchronizedCAS 来操作,整个看起来就像是优化过且线程安全的HashMap,虽然在JDK1.8中还能看到Segment的数据结构,但是已经简化了属性,只是为了兼容旧版本。

其他内容跟HashMap大同小异,看一下与HashMap不同的地方。

TreeBin

TreeBin 从字面含义中可以理解为存储树形结构的容器,而树形结构就是指 TreeNode,所以 TreeBin 就是封装 TreeNode 的容器,它提供转换黑红树的一些条件和锁的控制:

 1TreeBin(TreeNode<K,V> b) {
 2    super(TREEBIN, null, null, null);
 3    this.first = b;
 4    TreeNode<K,V> r = null;
 5    for (TreeNode<K,V> x = b, next; x != null; x = next) {
 6        next = (TreeNode<K,V>)x.next;
 7        x.left = x.right = null;
 8        if (r == null) {
 9            x.parent = null;
10            x.red = false;
11            r = x;
12        }
13        else {
14            K k = x.key;
15            int h = x.hash;
16            Class<?> kc = null;
17            for (TreeNode<K,V> p = r;;) {
18                int dir, ph;
19                K pk = p.key;
20                if ((ph = p.hash) > h)
21                    dir = -1;
22                else if (ph < h)
23                    dir = 1;
24                else if ((kc == null &&
25                          (kc = comparableClassFor(k)) == null) ||
26                         (dir = compareComparables(kc, k, pk)) == 0)
27                    dir = tieBreakOrder(k, pk);
28                TreeNode<K,V> xp = p;
29                if ((p = (dir <= 0) ? p.left : p.right) == null) {
30                    x.parent = xp;
31                    if (dir <= 0)
32                        xp.left = x;
33                    else
34                        xp.right = x;
35                    r = balanceInsertion(r, x);
36                    break;
37                }
38            }
39        }
40    }
41    this.root = r;
42    assert checkInvariants(root);
43}

初始化

1// 初始化 ConcurrentHashMap
2ConcurrentHashMap<String, String> map =  new ConcurrentHashMap();  

其中无参构造函数:

1public ConcurrentHashMap() {}

从上面的无参构造函数 可以看到,ConcurrentHashMap 的初始化其实是一个空实现,里面并没有做任何事,这也是和其他的集合类有区别的地方,初始化操作并不是在构造函数实现的,而是在 put 操作中实现,当然 ConcurrentHashMap 还提供了其他的构造函数,有指定容量大小或者指定负载因子,跟HashMap一样。

添加元素

 1public V put(K key, V value) {
 2    return putVal(key, value, false);
 3}
 4
 5/** Implementation for put and putIfAbsent */
 6final V putVal(K key, V value, boolean onlyIfAbsent) {
 7    if (key == null || value == null) throw new NullPointerException();
 8    int hash = spread(key.hashCode());
 9    int binCount = 0;
10    for (Node<K,V>[] tab = table;;) {
11        Node<K,V> f; int n, i, fh;
12        if (tab == null || (n = tab.length) == 0)
13            tab = initTable();
14        else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
15            if (casTabAt(tab, i, null,     
16                         new Node<K,V>(hash, key, value, null)))
17                break;                   // no lock when adding to empty bin
18        }
19        else if ((fh = f.hash) == MOVED)
20            tab = helpTransfer(tab, f);
21        else {
22            V oldVal = null;
23            synchronized (f) { 
24                if (tabAt(tab, i) == f) {
25                    if (fh >= 0) {
26                        binCount = 1;
27                        for (Node<K,V> e = f;; ++binCount) {
28                            K ek;
29                            if (e.hash == hash &&
30                                ((ek = e.key) == key ||
31                                 (ek != null && key.equals(ek)))) {
32                                oldVal = e.val;
33                                if (!onlyIfAbsent)
34                                    e.val = value;
35                                break;
36                            }
37                            Node<K,V> pred = e;
38                            if ((e = e.next) == null) {
39                                pred.next = new Node<K,V>(hash, key,
40                                                          value, null);
41                                break;
42                            }
43                        }
44                    }
45                    else if (f instanceof TreeBin) {
46                        Node<K,V> p;
47                        binCount = 2;
48                        if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
49                                                              value)) != null) {
50                            oldVal = p.val;
51                            if (!onlyIfAbsent)
52                                p.val = value;
53                        }
54                    }
55                }
56            }
57            if (binCount != 0) {
58                if (binCount >= TREEIFY_THRESHOLD)
59                    treeifyBin(tab, i);
60                if (oldVal != null)
61                    return oldVal;
62                break;
63            }
64        }
65    }
66    addCount(1L, binCount);
67    return null;
68}

这个 put 的过程很清晰,对当前的 table 进行无条件自循环直到 put 成功,可以分成以下六步流程:

  1. 如果没有初始化就先调用 initTable() 方法来进行初始化;
  2. 如果没有hash冲突就直接CAS插入;
  3. 如果还在进行扩容操作就先进行扩容;
  4. 如果存在 hash 冲突,就加锁来保证线程安全,这里有两种情况,一种是链表形式就直接遍历到尾端插入,一种是红黑树就按照红黑树结构插入;
  5. 最后一个如果 hash 冲突时会形成 Node 链表,在链表长度超过8,Node 数组超过64时会将链表结构转换为红黑树的结构,break再一次进入循环;
  6. 如果添加成功就调用 addCount() 方法统计size,并且检查是否需要扩容。

现在来对每一步的细节进行源码分析,在第一步中,符合条件会进行初始化操作,我们来看看 initTable() 方法:

 1private final Node<K,V>[] initTable() {
 2    Node<K,V>[] tab; int sc;
 3    while ((tab = table) == null || tab.length == 0) {
 4        if ((sc = sizeCtl) < 0)
 5            Thread.yield(); // lost initialization race; just spin
 6        else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
 7            try {
 8                if ((tab = table) == null || tab.length == 0) {
 9                    int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
10                    @SuppressWarnings("unchecked")
11                    Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
12                    table = tab = nt;
13                    sc = n - (n >>> 2);
14                }
15            } finally {
16                sizeCtl = sc;
17            }
18            break;
19        }
20    }
21    return tab;
22}

在第二步中没有 hash 冲突就直接调用Unsafe的方法CAS插入该元素,进入第三步如果容器正在扩容,则会调用helpTransfer() 方法帮助扩容,现在我们跟进 helpTransfer() 方法看看:

 1final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
 2    Node<K,V>[] nextTab; int sc;
 3    if (tab != null && (f instanceof ForwardingNode) &&
 4        (nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) {
 5        int rs = resizeStamp(tab.length);
 6        while (nextTab == nextTable && table == tab &&
 7               (sc = sizeCtl) < 0) {
 8            if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
 9                sc == rs + MAX_RESIZERS || transferIndex <= 0)
10                break;
11            if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
12                transfer(tab, nextTab);
13                break;
14            }
15        }
16        return nextTab;
17    }
18    return table;
19}

其实 helpTransfer() 方法的目的就是调用多个工作线程一起帮助进行扩容,这样的效率就会更高,而不是只有检查到要扩容的那个线程进行扩容操作,其他线程等待扩容操作完成才能工作,既然这里涉及到扩容的操作,我们也一起来看看扩容方法 transfer()

  1private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
  2    int n = tab.length, stride;
  3    if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
  4        stride = MIN_TRANSFER_STRIDE; // subdivide range
  5    if (nextTab == null) {            // initiating
  6        try {
  7            @SuppressWarnings("unchecked")
  8            Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
  9            nextTab = nt;
 10        } catch (Throwable ex) {      // try to cope with OOME
 11            sizeCtl = Integer.MAX_VALUE;
 12            return;
 13        }
 14        nextTable = nextTab;
 15        transferIndex = n;
 16    }
 17    int nextn = nextTab.length;
 18    ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
 19    boolean advance = true;
 20    boolean finishing = false; // to ensure sweep before committing nextTab
 21    for (int i = 0, bound = 0;;) {
 22        Node<K,V> f; int fh;
 23        while (advance) {
 24            int nextIndex, nextBound;
 25            if (--i >= bound || finishing)
 26                advance = false;
 27            else if ((nextIndex = transferIndex) <= 0) {
 28                i = -1;
 29                advance = false;
 30            }
 31            else if (U.compareAndSwapInt
 32                     (this, TRANSFERINDEX, nextIndex,
 33                      nextBound = (nextIndex > stride ?
 34                                   nextIndex - stride : 0))) {
 35                bound = nextBound;
 36                i = nextIndex - 1;
 37                advance = false;
 38            }
 39        }
 40        if (i < 0 || i >= n || i + n >= nextn) {
 41            int sc;
 42            if (finishing) {
 43                nextTable = null;
 44                table = nextTab;
 45                sizeCtl = (n << 1) - (n >>> 1);
 46                return;
 47            }
 48            if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
 49                if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
 50                    return;
 51                finishing = advance = true;
 52                i = n; // recheck before commit
 53            }
 54        }
 55        else if ((f = tabAt(tab, i)) == null)
 56            advance = casTabAt(tab, i, null, fwd);
 57        else if ((fh = f.hash) == MOVED)
 58            advance = true; // already processed
 59        else {
 60            synchronized (f) {
 61                if (tabAt(tab, i) == f) {
 62                    Node<K,V> ln, hn;
 63                    if (fh >= 0) {
 64                        int runBit = fh & n;
 65                        Node<K,V> lastRun = f;
 66                        for (Node<K,V> p = f.next; p != null; p = p.next) {
 67                            int b = p.hash & n;
 68                            if (b != runBit) {
 69                                runBit = b;
 70                                lastRun = p;
 71                            }
 72                        }
 73                        if (runBit == 0) {
 74                            ln = lastRun;
 75                            hn = null;
 76                        }
 77                        else {
 78                            hn = lastRun;
 79                            ln = null;
 80                        }
 81                        for (Node<K,V> p = f; p != lastRun; p = p.next) {
 82                            int ph = p.hash; K pk = p.key; V pv = p.val;
 83                            if ((ph & n) == 0)
 84                                ln = new Node<K,V>(ph, pk, pv, ln);
 85                            else
 86                                hn = new Node<K,V>(ph, pk, pv, hn);
 87                        }
 88                        setTabAt(nextTab, i, ln);
 89                        setTabAt(nextTab, i + n, hn);
 90                        setTabAt(tab, i, fwd);
 91                        advance = true;
 92                    }
 93                    else if (f instanceof TreeBin) {
 94                        TreeBin<K,V> t = (TreeBin<K,V>)f;
 95                        TreeNode<K,V> lo = null, loTail = null;
 96                        TreeNode<K,V> hi = null, hiTail = null;
 97                        int lc = 0, hc = 0;
 98                        for (Node<K,V> e = t.first; e != null; e = e.next) {
 99                            int h = e.hash;
100                            TreeNode<K,V> p = new TreeNode<K,V>
101                                (h, e.key, e.val, null, null);
102                            if ((h & n) == 0) {
103                                if ((p.prev = loTail) == null)
104                                    lo = p;
105                                else
106                                    loTail.next = p;
107                                loTail = p;
108                                ++lc;
109                            }
110                            else {
111                                if ((p.prev = hiTail) == null)
112                                    hi = p;
113                                else
114                                    hiTail.next = p;
115                                hiTail = p;
116                                ++hc;
117                            }
118                        }
119                        ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) :
120                        (hc != 0) ? new TreeBin<K,V>(lo) : t;
121                        hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) :
122                        (lc != 0) ? new TreeBin<K,V>(hi) : t;
123                        setTabAt(nextTab, i, ln);
124                        setTabAt(nextTab, i + n, hn);
125                        setTabAt(tab, i, fwd);
126                        advance = true;
127                    }
128                }
129            }
130        }
131    }
132}

扩容过程有点复杂,这里主要涉及到多线程并发扩容,ForwardingNode 的作用就是支持扩容操作,将已处理的节点和空节点置为 ForwardingNode,并发处理时多个线程经过ForwardingNode就表示已经遍历了,就往后遍历。

介绍完扩容过程,再次回到 put 流程..

在第四步是向链表或者红黑树里加节点

第五步调用 treeifyBin() 方法进行链表转红黑树的过程

 1private final void treeifyBin(Node<K,V>[] tab, int index) {
 2    Node<K,V> b; int n, sc;
 3    if (tab != null) {
 4        if ((n = tab.length) < MIN_TREEIFY_CAPACITY)
 5            tryPresize(n << 1);
 6        else if ((b = tabAt(tab, index)) != null && b.hash >= 0) {
 7            synchronized (b) {
 8                if (tabAt(tab, index) == b) {
 9                    TreeNode<K,V> hd = null, tl = null;
10                    for (Node<K,V> e = b; e != null; e = e.next) {
11                        TreeNode<K,V> p =
12                            new TreeNode<K,V>(e.hash, e.key, e.val,
13                                              null, null);
14                        if ((p.prev = tl) == null)
15                            hd = p;
16                        else
17                            tl.next = p;
18                        tl = p;
19                    }
20                    setTabAt(tab, index, new TreeBin<K,V>(hd));
21                }
22            }
23        }
24    }
25}

到第六步表示已经数据加入成功了,现在调用 addCount() 方法计算 ConcurrentHashMap 的 size,在原来的基础上加一,现在来看看 addCount() 方法

 1private final void addCount(long x, int check) {
 2    CounterCell[] as; long b, s;
 3    if ((as = counterCells) != null ||
 4        !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
 5        CounterCell a; long v; int m;
 6        boolean uncontended = true;
 7        if (as == null || (m = as.length - 1) < 0 ||
 8            (a = as[ThreadLocalRandom.getProbe() & m]) == null ||
 9            !(uncontended =
10              U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
11            fullAddCount(x, uncontended);
12            return;
13        }
14        if (check <= 1)
15            return;
16        s = sumCount();
17    }
18    if (check >= 0) {
19        Node<K,V>[] tab, nt; int n, sc;
20        while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
21               (n = tab.length) < MAXIMUM_CAPACITY) {
22            int rs = resizeStamp(n);
23            if (sc < 0) {
24                if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
25                    sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
26                    transferIndex <= 0)
27                    break;
28                if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
29                    transfer(tab, nt);
30            }
31            else if (U.compareAndSwapInt(this, SIZECTL, sc,
32                                         (rs << RESIZE_STAMP_SHIFT) + 2))
33                transfer(tab, null);
34            s = sumCount();
35        }
36    }
37}

put 的流程现在已经分析完了,可以发现,它在并发处理中使用的是乐观锁,当有冲突的时候才进行并发处理,而且流程步骤很清晰,但是细节设计的很复杂,毕竟多线程的场景也复杂。

获取元素

 1public V get(Object key) {
 2    Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
 3    int h = spread(key.hashCode());
 4    if ((tab = table) != null && (n = tab.length) > 0 &&
 5        (e = tabAt(tab, (n - 1) & h)) != null) {
 6        if ((eh = e.hash) == h) {
 7            if ((ek = e.key) == key || (ek != null && key.equals(ek)))
 8                return e.val;
 9        }
10        else if (eh < 0)
11            return (p = e.find(h, key)) != null ? p.val : null;
12        while ((e = e.next) != null) {
13            if (e.hash == h &&
14                ((ek = e.key) == key || (ek != null && key.equals(ek))))
15                return e.val;
16        }
17    }
18    return null;
19}

ConcurrentHashMap 的 get 操作的流程很简单,也很清晰,可以分为三个步骤来描述

  1. 计算hash值,定位到该table索引位置,如果是首节点符合就返回
  2. 如果遇到扩容的时候,会调用标志正在扩容节点ForwardingNode的find方法,查找该节点,匹配就返回
  3. 以上都不符合的话,就往下遍历节点,匹配就返回,否则最后就返回null

size()方法

 1public int size() {
 2    long n = sumCount();
 3    return ((n < 0L) ? 0 :
 4            (n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :
 5            (int)n);
 6}
 7
 8final long sumCount() {
 9    CounterCell[] as = counterCells; CounterCell a;
10    long sum = baseCount;
11    if (as != null) {
12        for (int i = 0; i < as.length; ++i) {
13            if ((a = as[i]) != null)
14                sum += a.value;
15        }
16    }
17    return sum;
18}

在 JDK1.8 版本中,对于 size 的计算,在扩容和addCount()方法就已经有处理了,JDK1.7是在调用size()方法才去计算,其实在并发集合中去计算size是没有多大的意义的,因为size是实时在变的,只能计算某一刻的大小,但是某一刻太快了,人的感知是一个时间段,所以并不是很精确。

总结

其实可以看出JDK1.8版本的 ConcurrentHashMap 的数据结构已经接近HashMap,相对而言,ConcurrentHashMap只是增加了同步的操作来控制并发,从JDK1.7版本的ReentrantLock+Segment+HashEntry,到JDK1.8版本中synchronized+CAS+HashEntry+红黑树,相对而言,总结如下思考:

  1. JDK1.8 的实现降低锁的粒度,JDK1.7版本锁的粒度是基于Segment的,包含多个HashEntry,而JDK1.8锁的粒度就是HashEntry(首节点)

  2. JDK1.8 版本的数据结构变得更加简单,使得操作也更加清晰流畅,因为已经使用 synchronized 来进行同步,所以不需要分段锁的概念,也就不需要 Segment 这种数据结构了,由于粒度的降低,实现的复杂度也增加了

  3. JDK1.8 使用红黑树来优化链表,基于长度很长的链表的遍历是一个很漫长的过程,而红黑树的遍历效率是很快的,代替一定阈值的链表,这样形成一个最佳拍档

  4. JDK1.8 为什么使用内置锁 synchronized 来代替重入锁 ReentrantLock,我觉得有以下几点:

    • 因为粒度降低了。在粗粒度加锁中 ReentrantLock 可能通过 Condition 来控制各个低粒度的边界,更加的灵活;低粒度加锁方式下,Condition的优势就没有了,因此 synchronized 并不比 ReentrantLock 差;
    1. JVM 的开发团队从来都没有放弃 synchronized,而且基于 JVM 的 synchronized 优化空间更大,使用内嵌的关键字比使用 API 更加自然;
    2. 在大量的数据操作下,对于 JVM 的内存压力,基于 API 的 ReentrantLock 会开销更多的内存,虽然不是瓶颈,但是也是一个选择依据。