本文共 12731 字,大约阅读时间需要 42 分钟。
上篇内容大概介绍了ConcurrentHashMap在看源码时,需要了解的一些知识概念,整个数据结构以及里面常用的属性含义。
参见
这篇我们从它比较基础的操作开始分析
初始化操作
对于一个新创建的ConcurrentHashMap,我们会给一个table为空的数组,也就是说我们并不会去初始化等,等到需要put元素的时候,回去校验table是否初始化,如果没有初始化,我们会进行一个初始化操作。
private final Node[] initTable() { Node [] tab; int sc; while ((tab = table) == null || tab.length == 0) { if ((sc = sizeCtl) < 0) Thread.yield(); // 如果当前sizeCtl已经小于0,说明已经有线程在初始化了,当前线程先进行休眠 else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) { try { if ((tab = table) == null || tab.length == 0) { //将sizeCtl更新成功之后,再次判断是否为null int n = (sc > 0) ? sc : DEFAULT_CAPACITY; @SuppressWarnings("unchecked") Node [] nt = (Node [])new Node [n]; table = tab = nt; sc = n - (n >>> 2); } } finally { sizeCtl = sc; //如果初始化成功,将数组长度-数组长度一般作为size } break; } } return tab;}
初始化table时,我们会对sizeCtl 进行一个判断,如果他的值是小于0,要么是正在进行扩容,要么是在初始化,可以判断当前的结果最大可能是 -1 ,说明有其他线程在对其进行初始化操作,那么当前线程主动让出,
如果sizeCtl的值是0,说明未初始化,通过unsafe方式来将其改为 -1 ,如果修改成功,创建一个长度为n的数组(如果sizeCtrl大于0即已经知道扩容大小,否则使用默认长度16)
最后返回创建的tab数组
static finalboolean casTabAt(Node [] tab, int i, Node c, Node v) { return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v);}
接下来可以看到如果添加的元素,在数组中没有,那么我们直接使用unsafe的这种原子操作来添加头结点元素,并且此操作不需要添加锁。
如果说当前索引处是一个转移结点(forwardingNode)说明正在进行扩容操作,那么当前线程会添加进去帮助扩容
final Node[] helpTransfer(Node [] tab, Node f) { Node [] nextTab; int sc; // 如果 table 不是空 且 node 节点是转移类型,数据检验 // 且 node 节点的 nextTable(新 table) 不是空,同样也是数据校验 // 尝试帮助扩容 if (tab != null && (f instanceof ForwardingNode) && (nextTab = ((ForwardingNode )f).nextTable) != null) { // 根据 length 得到一个标识符号 int rs = resizeStamp(tab.length); // 如果 nextTab 没有被并发修改 且 tab 也没有被并发修改 // 且 sizeCtl < 0 (说明还在扩容) while (nextTab == nextTable && table == tab && (sc = sizeCtl) < 0) { // 如果 sizeCtl 无符号右移 16 不等于 rs ( sc前 16 位如果不等于标识符,则标识符变化了) // 或者 sizeCtl == rs + 1 (扩容结束了,不再有线程进行扩容)(默认第一个线程设置 sc ==rs 左移 16 位 + 2,当第一个线程结束扩容了,就会将 sc 减一。这个时候,sc 就等于 rs + 1) // 或者 sizeCtl == rs + 65535 (如果达到最大帮助线程的数量,即 65535) // 或者转移下标正在调整 (扩容结束) // 结束循环,返回 table if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 || sc == rs + MAX_RESIZERS || transferIndex <= 0) break; // 如果以上都不是, 将 sizeCtl + 1, (表示增加了一个线程帮助其扩容) if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) { // 进行转移 transfer(tab, nextTab); // 结束循环 break; } } return nextTab; } return table;}
else { V oldVal = null; synchronized (f) { if (tabAt(tab, i) == f) { if (fh >= 0) { binCount = 1; for (Nodee = f;; ++binCount) { K ek; if (e.hash == hash && ((ek = e.key) == key || (ek != null && key.equals(ek)))) { //如果key相同,认为插入的是相同的元素,直接更新 oldVal = e.val; if (!onlyIfAbsent) e.val = value; break; } Node pred = e;
if ((e = e.next) == null) { //遍历到链表最后一个元素,创建一个新的节点 pred.next = new Node(hash, key, value, null); break; } } } else if (f instanceof TreeBin) { Node p; binCount = 2; if ((p = ((TreeBin )f).putTreeVal(hash, key, value)) != null) { oldVal = p.val; if (!onlyIfAbsent) p.val = value; } } } }
如果当前节点还没有进行扩容,那么先把节点第一个元素添加锁,锁定之后,进行判断
如果f等于tab表中索引为i元素,即未被其他线程修改,
接下来,如果hash值大于0,遍历这个链表,查找当前链表中是否有key的值与当前新添加的值相同,如果相同,任务添加的元素已经存在,进行替换。
如果新添加的元素不存在,那么在链表的最后添加一个新的节点,如果链表类型是TreeBin,红黑树,那么就直接在红黑树中添加元素,这个对于红黑树中添加元素也是先判断当前插入元素是否存在,如果不存在插入新元素,并且调整红黑树结构。
如果元素添加成功之后,会先判断链表长度是否超过8,如果超过8,将链表转为TreeBin
添加元素之后,会考虑进行扩容
private final void addCount(long x, int check) { CounterCell[] as; long b, s; 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))) { fullAddCount(x, uncontended); return; } if (check <= 1) return; s = sumCount(); } if (check >= 0) { Node[] tab, nt; int n, sc; while (s >= (long)(sc = sizeCtl) && (tab = table) != null && (n = tab.length) < MAXIMUM_CAPACITY) { int rs = resizeStamp(n); if (sc < 0) { if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 || sc == rs + MAX_RESIZERS || (nt = nextTable) == null || transferIndex <= 0) //这个条件判断和上面协助扩容中的条件判断一样,如果已经扩容结束,直接break; break; if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) transfer(tab, nt); //sc小于0说明正在扩容,线程数加 1 } else if (U.compareAndSwapInt(this, SIZECTL, sc, (rs << RESIZE_STAMP_SHIFT) + 2)) transfer(tab, null); s = sumCount(); } } }
下面是扩容的方法
private final void transfer(Node[] tab, Node [] nextTab) { int n = tab.length, stride; if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE) stride = MIN_TRANSFER_STRIDE; // subdivide range if (nextTab == null) { // initiating try { @SuppressWarnings("unchecked") Node [] nt = (Node [])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; ForwardingNode fwd = new ForwardingNode (nextTab); boolean advance = true; boolean finishing = false; // to ensure sweep before committing nextTab for (int i = 0, bound = 0;;) { Node 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); 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) 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 ln, hn; if (fh >= 0) { int runBit = fh & n; Node lastRun = f; for (Node p = f.next; p != null; p = p.next) { int b = p.hash & n; if (b != runBit) { runBit = b; lastRun = p; } } if (runBit == 0) { ln = lastRun; hn = null; } else { hn = lastRun; ln = null; } for (Node p = f; p != lastRun; p = p.next) { int ph = p.hash; K pk = p.key; V pv = p.val; if ((ph & n) == 0) ln = new Node (ph, pk, pv, ln); else hn = new Node (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 t = (TreeBin )f; TreeNode lo = null, loTail = null; TreeNode hi = null, hiTail = null; int lc = 0, hc = 0; for (Node e = t.first; e != null; e = e.next) { int h = e.hash; TreeNode p = new TreeNode (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; } } ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) : (hc != 0) ? new TreeBin (lo) : t; hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) : (lc != 0) ? new TreeBin (hi) : t; setTabAt(nextTab, i, ln); setTabAt(nextTab, i + n, hn); setTabAt(tab, i, fwd); advance = true; } } } } } }
方法很长,我们来总结一下:
1、在原来数组的长度的基础上,创建一个长度是其二倍的数组
2、根据新创建的table,创建一个ForwardingNode节点
3、接下来去遍历table中每个位置上的元素,比如说从索引14开始,如果索引14的位置上没有元素,我们放置一个forwardingnNode来记录,这个索引位置我已经访问过了,等你来访问的时候,可以直接跳过了
4、如果说你看到索引14的位置上是moved,说明这个位置上的元素,已经处理过了,你不需要再管了
5、如果当前索引处有元素,需要将这个链表或者是红黑树移动到新的数组当中去,那么就先把当前位置上的对象加锁。
如果是链表的话,计算每个链表下的元素的key.hash & length == 0 这个当时在介绍hashMap的时候说过,key.hash&length==0 表示扩容后,该元素的索引位置不会发生变化,如果等于1 ,说明扩容后,该元素的索引会调整至当前索引+length的位置, 所以以上关于链表的操作就是将原来的链表拆分成两个不同的链表,分别放在了扩容后i的位置和i+n的位置。
同时将原tab中索引i的位置使用forwardingNode记录,表示该索引下的元素已经遍历过了
同样,对于红黑树,也是这样的操作,记录扩容后,原来的红黑树如何变化,同时会判断扩容后,红黑树的长度是否小于6,如果小于6转化成链表,并记录原tab 索引处node为forwardingNode
关于Node转TreeBin的方法
private final void treeifyBin(Node[] tab, int index) { Node b; int n, sc; if (tab != null) { if ((n = tab.length) < MIN_TREEIFY_CAPACITY) tryPresize(n << 1); else if ((b = tabAt(tab, index)) != null && b.hash >= 0) { synchronized (b) { if (tabAt(tab, index) == b) { TreeNode hd = null, tl = null; for (Node e = b; e != null; e = e.next) { TreeNode p = new TreeNode (e.hash, e.key, e.val, null, null); if ((p.prev = tl) == null) hd = p; else tl.next = p; tl = p; } setTabAt(tab, index, new TreeBin (hd)); } } } } }
上面这段代码,将原来的Node单向链表,生成了一个双向的TreeNode对象,还没生成红黑树
是在TreeBin(hd)中,拿到这个TreeNode对象,将其生成红黑树
TreeBin(TreeNodeb) { super(TREEBIN, null, null, null); this.first = b; TreeNode r = null; for (TreeNode x = b, next; x != null; x = next) { next = (TreeNode )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 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 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); }
每插入一个红黑树节点,都会重新调整红黑树