源码解析
put办法源码
public V put(K key, V value) { return putVal(key, value, false);}/** Implementation for put and putIfAbsent */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;//用来记录链表的长度 for (Node<K,V>[] tab = table;;) {//这里其实就是自旋操作，当呈现线程竞争时一直自旋 Node<K,V> f; int n, i, fh; if (tab == null || (n = tab.length) == 0)//如果数组为空，则进行数组初始化 tab = initTable();//初始化数组 // 通过hash值对应的数组下标失去第一个节点; 以volatile读的形式来读取table数组中的元素，保障每次拿到的数据都是最新的 else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) { //如果该下标返回的节点为空，则间接通过cas将新的值封装成node插入即可；如果cas失败，阐明存在竞争，则进入下一次循环 if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value, null))) break; // no lock when adding to empty bin } else if ((fh = f.hash) == MOVED) tab = helpTransfer(tab, f);//帮助扩容 else {//进入到这个分支，阐明f是以后nodes数组对应地位节点的头节点，并且不为空 V oldVal = null; synchronized (f) {//给对应的头结点加锁 if (tabAt(tab, i) == f) {//再次判断对应下标地位是否为f节点 if (fh >= 0) {//头结点的hash值大于0，阐明是链表 binCount = 1;//用来记录链表的长度 for (Node<K,V> e = f;; ++binCount) {//遍历链表 K ek; //如果发现雷同的key，则判断是否须要进行值的笼罩 if (e.hash == hash && ((ek = e.key) == key || (ek != null && key.equals(ek)))) { oldVal = e.val; if (!onlyIfAbsent)//默认状况下，间接笼罩旧的值 e.val = value; break; } //始终遍历到链表的最末端，间接把新的值退出到链表的最初面 Node<K,V> pred = e; if ((e = e.next) == null) { pred.next = new Node<K,V>(hash, key, value, null); break; } } } //如果以后的f节点是一颗红黑树 else if (f instanceof TreeBin) { Node<K,V> p; binCount = 2; //则调用红黑树的插入方法插入新的值 if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key, value)) != null) { oldVal = p.val;//同样，如果值曾经存在，则间接替换 if (!onlyIfAbsent) p.val = value; } } } } if (binCount != 0) {//阐明下面在做链表操作 //如果链表长度曾经达到临界值8 就须要把链表转换为树结构 if (binCount >= TREEIFY_THRESHOLD) treeifyBin(tab, i); if (oldVal != null)//如果val是被替换的，则返回替换之前的值 return oldVal; break; } } } //将以后ConcurrentHashMap的元素数量加1，有可能触发transfer操作(扩容) addCount(1L, binCount); return null;}
spread计算Hash值
static final int spread(int h) { return (h ^ (h >>> 16)) & HASH_BITS;}
initTable初始化表格
private final Node<K,V>[] initTable() { Node<K,V>[] tab; int sc; while ((tab = table) == null || tab.length == 0) { if ((sc = sizeCtl) < 0) //被其余线程抢占了初始化的操作,则间接让出本人的CPU工夫片 Thread.yield(); // lost initialization race; just spin else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) { //通过cas操作，将sizeCtl替换为-1，标识以后线程抢占到了初始化资格 try { if ((tab = table) == null || tab.length == 0) { int n = (sc > 0) ? sc : DEFAULT_CAPACITY;//默认初始容量为16 @SuppressWarnings("unchecked") //初始化数组，长度为16，或者初始化在结构ConcurrentHashMap的时候传入的长度 Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n]; table = tab = nt;//将这个数组赋值给table sc = n - (n >>> 2);//计算下次扩容的大小，理论就是以后容量的0.75倍，这里应用了右移来计算 } } finally { sizeCtl = sc;//设置sizeCtl为sc, 如果默认是16的话，那么这个时候sc=16*0.75=12 } break; } } return tab;}
tabAt计算下标
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);}
addCount计数
private final void addCount(long x, int check) { CounterCell[] as; long b, s; //判断counterCells是否为空， // 1. 如果为空，就通过cas操作尝试批改baseCount变量，对这个变量进行原子累加操作(做这个操作的意义是：如果在没有竞争的状况下，依然采纳baseCount来记录元素个数) // 2. 如果cas失败阐明存在竞争，这个时候不能再采纳baseCount来累加，而是通过CounterCell来记录 if ((as = counterCells) != null || !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) { CounterCell a; long v; int m; boolean uncontended = true;//是否抵触标识，默认为没有抵触 //这里有几个判断 // 1. 计数表为空则间接调用fullAddCount // 2. 从计数表中随机取出一个数组的地位为空，间接调用fullAddCount // 3. 通过CAS批改CounterCell随机地位的值，如果批改失败阐明呈现并发状况（这里又用到了一种奇妙的办法）， // 调用fullAndCount Random在线程并发的时候会有性能问题以及可能会产生雷同的随机数,ThreadLocalRandom.getProbe能够解决这个问题，并且性能要比Random高 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)//链表长度小于等于1，不须要思考扩容 return; s = sumCount();//统计ConcurrentHashMap元素个数 } if (check >= 0) {//如果binCount>=0，标识须要查看扩容 Node<K,V>[] tab, nt; int n, sc; //s标识汇合大小，如果汇合大小大于或等于扩容阈值（默认值的0.75） // 并且table不为空并且table的长度小于最大容量 while (s >= (long)(sc = sizeCtl) && (tab = table) != null && (n = tab.length) < MAXIMUM_CAPACITY) { int rs = resizeStamp(n);//这里是生成一个惟一的扩容戳 if (sc < 0) { //sc<0，也就是sizeCtl<0，阐明曾经有别的线程正在扩容了 // 这5个条件只有有一个条件为true，阐明以后线程不能帮忙进行此次的扩容，间接跳出循环 // sc >>> RESIZE_STAMP_SHIFT!=rs 示意比拟高RESIZE_STAMP_BITS位生成戳和rs是否相等，雷同 // sc=rs+1 示意扩容完结 // sc==rs+MAX_RESIZERS 示意帮忙线程线程曾经达到最大值了 // nt=nextTable -> 示意扩容曾经完结 // transferIndex<=0 示意所有的transfer工作都被支付完了，没有残余的hash桶来给本人本人好这个线程来做transfer if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 || sc == rs + MAX_RESIZERS || (nt = nextTable) == null || transferIndex <= 0) break; if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))//以后线程尝试帮忙此次扩容，如果胜利，则调用transfer transfer(tab, nt); } //如果以后没有在扩容，那么 rs 必定是一个负数，通过 rs <<RESIZE_STAMP_SHIFT 将 sc 设置 // 为一个正数， 2 示意有一个线程在执行扩容 else if (U.compareAndSwapInt(this, SIZECTL, sc, (rs << RESIZE_STAMP_SHIFT) + 2)) transfer(tab, null); s = sumCount();// 从新计数，判断是否须要开启下一轮扩容 } }}
sumCount计数
final long sumCount() { CounterCell[] as = counterCells; CounterCell a; long sum = baseCount; if (as != null) { for (int i = 0; i < as.length; ++i) { if ((a = as[i]) != null) sum += a.value; } } return sum;}
fullAddCount源码剖析
private final void fullAddCount(long x, boolean wasUncontended) { int h; //获取以后线程的probe的值，如果值为0，则初始化以后线程的probe的值,probe就是随机数 if ((h = ThreadLocalRandom.getProbe()) == 0) { ThreadLocalRandom.localInit(); // force initialization h = ThreadLocalRandom.getProbe(); wasUncontended = true;// 因为从新生成了probe，未抵触标记位设置为true } boolean collide = false; // True if last slot nonempty for (;;) { CounterCell[] as; CounterCell a; int n; long v; //阐明counterCells曾经被初始化过了 if ((as = counterCells) != null && (n = as.length) > 0) { if ((a = as[(n - 1) & h]) == null) {//通过该值与以后线程probe求与，取得cells的下标元素，和hash 表获取索引是一样的 if (cellsBusy == 0) { // Try to attach new Cell //cellsBusy=0示意counterCells不在初始化或者扩容状态下 CounterCell r = new CounterCell(x); // Optimistic create //结构一个CounterCell的值，传入元素个数 if (cellsBusy == 0 && //通过cas设置cellsBusy标识，避免其余线程来对counterCells并发解决 U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) { boolean created = false; try { // Recheck under lock CounterCell[] rs; int m, j; //将初始化的r对象的元素个数放在对应下标的地位 if ((rs = counterCells) != null && (m = rs.length) > 0 && rs[j = (m - 1) & h] == null) { rs[j] = r; created = true; } } finally {//复原标记位 cellsBusy = 0; } if (created)//创立胜利，退出循环 break; continue; // Slot is now non-empty //阐明指定cells下标地位的数据不为空，则进行下一次循环 } } collide = false; } //阐明在addCount办法中cas失败了，并且获取probe的值不为空 else if (!wasUncontended) // CAS already known to fail wasUncontended = true; // Continue after rehash//设置为未抵触标识，进入下一次自旋 else if (U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))//因为指定下标地位的cell值不为空，则间接通过cas进行原子累加，如果胜利，则间接退出 break; else if (counterCells != as || n >= NCPU)//如果曾经有其余线程建设了新的counterCells或者CounterCells大于CPU外围数（很奇妙，线程的并发数不会超过cpu外围数） collide = false; // At max size or stale //设置以后线程的循环失败不进行扩容 else if (!collide)//复原collide状态，标识下次循环会进行扩容 collide = true; else if (cellsBusy == 0 &&//进入这个步骤，阐明CounterCell数组容量不够，线程竞争较大，所以先设置一个标识示意为正在扩容 else U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) { try { if (counterCells == as) {// Expand table unless stale //扩容一倍 2变成4，这个扩容比较简单 CounterCell[] rs = new CounterCell[n << 1]; for (int i = 0; i < n; ++i) rs[i] = as[i]; counterCells = rs; } } finally { cellsBusy = 0;//复原标识 } collide = false; continue; // Retry with expanded table //持续下一次自旋 } h = ThreadLocalRandom.advanceProbe(h);//更新随机数的值 } else if (cellsBusy == 0 && counterCells == as && //cellsBusy=0示意没有在做初始化，通过cas更新cellsbusy的值标注以后线程正在做初始化操作 U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) { boolean init = false; try { // Initialize table if (counterCells == as) { CounterCell[] rs = new CounterCell[2];//初始化容量为2 rs[h & 1] = new CounterCell(x);//将x也就是元素的个数放在指定的数组下标地位 counterCells = rs;//赋值给counterCells init = true;//设置初始化实现标识 } } finally { cellsBusy = 0;//复原标识 } if (init) break; } else if (U.compareAndSwapLong(this, BASECOUNT, v = baseCount, v + x))//竞争强烈，其它线程占据cell 数组，间接累加在base变量中 break; // Fall back on using base }}
transfer扩容
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) { int n = tab.length, stride; //将 (n>>>3相当于 n/8) 而后除以 CPU外围数。如果失去的后果小于 16，那么就应用 16 // 这里的目标是让每个 CPU 解决的桶一样多，避免出现转移工作不平均的景象，如果桶较少的话，默认一个 CPU（一个线程）解决 16 个桶，也就是长度为16的时候，扩容的时候只会有一个线程来扩容 if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE) stride = MIN_TRANSFER_STRIDE; // subdivide range //nextTab未初始化，nextTab是用来扩容的node数组 if (nextTab == null) { // initiating try { @SuppressWarnings("unchecked") //新建一个n<<1原始table大小的nextTab,也就是32 Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1]; nextTab = nt;//赋值给nextTab } catch (Throwable ex) { // try to cope with OOME sizeCtl = Integer.MAX_VALUE;//扩容失败，sizeCtl应用int的最大值 return; } nextTable = nextTab;//更新成员变量 transferIndex = n;//更新转移下标，示意转移时的下标 } int nextn = nextTab.length;//新的tab的长度 // 创立一个 fwd 节点，示意一个正在被迁徙的Node，并且它的hash值为-1(MOVED)， // 也就是后面咱们在讲putval办法的时候，会有一个判断MOVED的逻辑。 // 它的作用是用来占位，示意原数组中地位i处的节点实现迁徙当前， // 就会在i地位设置一个fwd来通知其余线程这个地位曾经解决过了，具体后续还会在讲 ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab); // 首次推动为 true，如果等于 true，阐明须要再次推动一个下标（i--），反之，如果是 false，那么就不能推动下标，须要将以后的下标处理完毕能力持续推动 boolean advance = true; //判断是否曾经扩容实现，实现就return，退出循环 boolean finishing = false; // to ensure sweep before committing nextTab //通过for自循环解决每个槽位中的链表元素，默认advace为真，通过CAS设置transferIndex属性值， // 并初始化i和bound值，i指以后解决的槽位序号，bound指须要解决的槽位边界，先解决槽位15的节点； for (int i = 0, bound = 0;;) { // 这个循环应用CAS一直尝试为以后线程分配任务 // 直到调配胜利或工作队列曾经被全副调配结束 // 如果以后线程曾经被调配过bucket区域 // 那么会通过--i指向下一个待处理bucket而后退出该循环 Node<K,V> f; int fh; while (advance) { int nextIndex, nextBound; //--i示意下一个待处理的bucket，如果它>=bound,示意以后线程曾经调配过bucket区域 if (--i >= bound || finishing) advance = false; else if ((nextIndex = transferIndex) <= 0) {//示意所有bucket曾经被调配结束 i = -1; advance = false; } //通过cas来批改TRANSFERINDEX,为以后线程分配任务，解决的节点区间为(nextBound,nextIndex)->(0,15) else if (U.compareAndSwapInt (this, TRANSFERINDEX, nextIndex, nextBound = (nextIndex > stride ? nextIndex - stride : 0))) { bound = nextBound; i = nextIndex - 1; advance = false; } } //i<0阐明曾经遍历完旧的数组，也就是以后线程曾经解决完所有负责的bucket if (i < 0 || i >= n || i + n >= nextn) { int sc; if (finishing) {//如果实现了扩容 nextTable = null;//删除成员变量 table = nextTab;//更新table数组 sizeCtl = (n << 1) - (n >>> 1);//更新阈值(32*0.75=24) return; } // sizeCtl 在迁徙前会设置为 (rs << RESIZE_STAMP_SHIFT) + 2 // 而后，每减少一个线程参加迁徙就会将 sizeCtl 加 1， // 这里应用 CAS 操作对 sizeCtl 的低16位进行减 1，代表做完了属于本人的工作 if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) { //第一个扩容的线程，执行transfer办法之前，会设置 sizeCtl = (resizeStamp(n) << RESIZE_STAMP_SHIFT) + 2) // 后续帮其扩容的线程，执行transfer办法之前，会设置 sizeCtl = sizeCtl+1 每一个退出transfer的办法的线程，退出之前， // 会设置 sizeCtl = sizeCtl-1 那么最初一个线程退出时：必然有 sc == (resizeStamp(n) << RESIZE_STAMP_SHIFT) + 2)， // 即 (sc - 2) == resizeStamp(n) << RESIZE_STAMP_SHIFT // 如果 sc - 2 不等于标识符左移 16 位。如果他们相等了， // 阐明没有线程在帮忙他们扩容了。也就是说，扩容完结了。 if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT) return; // 如果相等，扩容完结了，更新 finising 变量 finishing = advance = true; // 再次循环检查一下整张表 i = n; // recheck before commit } } // 如果地位 i 处是空的，没有任何节点，那么放入刚刚初始化的 ForwardingNode ”空节点“ else if ((f = tabAt(tab, i)) == null) advance = casTabAt(tab, i, null, fwd); //示意该地位曾经实现了迁徙，也就是如果线程A曾经解决过这个节点，那么线程B解决这个节点时，hash值肯定为MOVED else if ((fh = f.hash) == MOVED) advance = true; // already processed else { synchronized (f) {//对数组该节点地位加锁，开始解决数组该地位的迁徙工作 if (tabAt(tab, i) == f) { Node<K,V> ln, hn;///ln 示意低位， hn 示意高位 接下来这段代码的作用 是把链表拆分成两局部， 0 在低位， 1 在高位 if (fh >= 0) { int runBit = fh & n; Node<K,V> lastRun = f; //遍历以后bucket的链表，目标是尽量重用Node链表尾部的一部分 for (Node<K,V> p = f.next; p != null; p = p.next) { int b = p.hash & n; if (b != runBit) { runBit = b; lastRun = p; } } if (runBit == 0) {//如果最初更新的runBit是0，设置低位节点 ln = lastRun; hn = null; } else {//否则，设置高位节点 hn = lastRun; ln = null; } //结构高位以及低位的链表 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) ln = new Node<K,V>(ph, pk, pv, ln); else hn = new Node<K,V>(ph, pk, pv, hn); } setTabAt(nextTab, i, ln);//将低位的链表放在 i 地位也就是不动 setTabAt(nextTab, i + n, hn);//将高位链表放在 i+n 地位 setTabAt(tab, i, fwd);// 把旧 table 的 hash 桶中搁置转发节点，表明此 hash 桶曾经被解决 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; } } 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; } } } } }}
resizeStamp
static final int resizeStamp(int n) { return Integer.numberOfLeadingZeros(n) | (1 << (RESIZE_STAMP_BITS - 1)); }
helpTransfer
private final void tryPresize(int size) { //对size进行修复,次要目标是避免传入的值不是一个2次幂的整数，而后通过tableSizeFor来讲入参转化为离该整数最近的2次幂 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; //上面这段代码和initTable是一样的，如果table没有初始化，则开始初始化 if (tab == null || (n = tab.length) == 0) { n = (sc > c) ? sc : c; if (U.compareAndSwapInt(this, SIZECTL, sc, -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; } } } else if (c <= sc || n >= MAXIMUM_CAPACITY) break; else if (tab == table) { //这段代码和addCount后局部代码是一样的，做辅助扩容操作 int rs = resizeStamp(n); 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; 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); } }}
treeifyBin
private final void treeifyBin(Node<K,V>[] tab, int index) { Node<K,V> b; int n, sc; if (tab != null) { if ((n = tab.length) < MIN_TREEIFY_CAPACITY)//tab的长度是不是小于64，如果是，则执行扩容 tryPresize(n << 1); else if ((b = tabAt(tab, index)) != null && b.hash >= 0) {//否则，将以后链表转化为红黑树结构存储 synchronized (b) { // 将链表转换成红黑树 if (tabAt(tab, index) == b) { TreeNode<K,V> hd = null, tl = null; 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) hd = p; else tl.next = p; tl = p; } setTabAt(tab, index, new TreeBin<K,V>(hd)); } } } }}
tryPresize
private final void tryPresize(int size) { //对size进行修复,次要目标是避免传入的值不是一个2次幂的整数，而后通过tableSizeFor来讲入参转化为离该整数最近的2次幂 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; //上面这段代码和initTable是一样的，如果table没有初始化，则开始初始化 if (tab == null || (n = tab.length) == 0) { n = (sc > c) ? sc : c; if (U.compareAndSwapInt(this, SIZECTL, sc, -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; } } } else if (c <= sc || n >= MAXIMUM_CAPACITY) break; else if (tab == table) { //这段代码和addCount后局部代码是一样的，做辅助扩容操作 int rs = resizeStamp(n); 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; 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); } }}