一、基本介绍

数据结构

ConcurrentHashMap的数据结构与HashMap基本类似。
区别在于：

  1.   1. 内部在数据写入时加了同步机制(CAS或synchronized)保证线程安全，读操作是无锁操作；
  2.   1. 并发执行扩容，扩容的效率更高。

hash值计算

concurrentHashMap：(h ^ (h >>> 16)) & HASH_BITS（HASH_BITS=01111111111111111111111111111111）
hashMap：h ^ (h >>> 16)

重要成员变量

sizeCtl

如果为负值，则表正在初始化或扩容
0：（默认值）底层数组还未初始化（table为null时，sizeCtl数值等于初始数组长度，默认值为0）
-1：底层数组正在初始化
-（1+扩容线程数）：进行扩容。
<-1：底层数组table正在扩容
>0：初始化完成后，sizeCtl保存下一次要扩容阈值（即table最大存放数据）。（例如初始容量16，则为12。因为扩容因子0.75。16*0.75=12）

MOVED

值为-1，正在扩容

MIN_TRANSFER_STRIDE

默认16, table扩容时, 每个线程最少迁移table的槽位个数。

TREEBIN

置为-2, 代表此元素后接红黑树。

nextTable

用于table迁移过程的临时数组, 在迁移过程中将元素全部迁移到nextTable上。

LOAD_FACTOR

负载因子, 默认0.75, 当table使用率达到75%时, 为减少table的hash碰撞, tabel长度将扩容一倍。

TREEIFY_THRESHOLD

默认8, 当链表长度大于8时, 链表转变为红黑树。

UNTREEIFY_THRESHOLD

默认6, 红黑树转变为链表的阈值。

transferIndex

当前已经迁移的元素下标，table容量从n扩到2n时, 按索引n到索引0的顺序开始迁移

ForwardingNode

一个特殊的Node节点, 其hashcode=MOVED, 代表此时table正在做扩容操作。扩容期间, 若table某个元素为null, 那么该元素设置为ForwardingNode, 当下个线程向这个元素插入数据时, 检查hashcode=MOVED, 就会帮着扩容。

ConcurrentHashMap由三部分构成, table+链表+红黑树,
table是一个数组, 既然是数组, 必须要在使用时确定数组的大小, 当table存放的元素过多时, 就需要扩容, 以减少碰撞发生次数, 扩容检查主要发生在插入元素(putVal())的过程:

  1.      - 一个线程插完元素后, 检查table使用率, 若超过阈值, 调用transfer进行扩容
  2.      - 一个线程插入数据时, 发现table对应元素的hash=MOVED, 那么调用helpTransfer()协助扩容。

put操作

1.普通场景添加。使用CAS操作，引入U = sun.misc.Unsafe.getUnsafe()类操作
2.添加过程中如果需要扩容，则将当前槽位hash值，设置为-1。当另外线程 添加的元素与当前槽位冲突时，则帮助扩容，
3.当出现hash冲突，即：出现链表时，使用synchronized加锁，锁加在hash槽位的第一个元素上。

源码分析

put操作

  1.     final V putVal(K key, V value, boolean onlyIfAbsent) {
  2.         if (key == null || value == null) throw new NullPointerException();
  3.         int hash = spread(key.hashCode());
  4.         int binCount = 0;
  5.         // for循环保证CAS操作成功后才退出。
  6.         for (Node<K,V>[] tab = table;;) {
  7.             Node<K,V> f; int n, i, fh;
  8.             if (tab == null || (n = tab.length) == 0)
  9.                 //只有在第一次put数据时，才会初始化底层数组
  10.                 tab = initTable();
  11.             // 判断是否存在冲突
  12.             // 利用U = sun.misc.Unsafe.getUnsafe()获取对应hash槽位元素，
  13.             else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
  14.                 //【不存在冲突】，则使用cas方式，添加新对象（解决hashMap中覆盖问题）
  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.             //执行到这，说明【存在hash冲突】，判断此槽位第一个元素hash值是否-1.（MOVED==-1表示正在进行扩容操作）
  20.             else if ((fh = f.hash) == MOVED)
  21.                 //当前线程帮助扩容；为当前线程至少分配16槽位处理，
  22.                 tab = helpTransfer(tab, f);
  23.             else {
  24.                 V oldVal = null;
  25.                 //锁加在hash槽位的第一个元素上.f表示槽位上第一个元素
  26.                 synchronized (f) {
  27.                     if (tabAt(tab, i) == f) {
  28.                         if (fh >= 0) {
  29.                             binCount = 1;
  30.                             for (Node<K,V> e = f;; ++binCount) {
  31.                                 K ek;
  32.                                 //判断新数据是否在链表是否存在，存在则覆盖。通过key判断。
  33.                                 if (e.hash == hash &&
  34.                                     ((ek = e.key) == key ||
  35.                                      (ek != null && key.equals(ek)))) {
  36.                                     oldVal = e.val;
  37.                                     if (!onlyIfAbsent)
  38.                                         e.val = value;
  39.                                     break;
  40.                                 }
  41.                                 //尾插法插入链表
  42.                                 Node<K,V> pred = e;
  43.                                 if ((e = e.next) == null) {
  44.                                     pred.next = new Node<K,V>(hash, key,
  45.                                                               value, null);
  46.                                     break;
  47.                                 }
  48.                             }
  49.                         }
  50.                         //是否为树结构，如果是，则加入树结构中
  51.                         else if (f instanceof TreeBin) {
  52.                             Node<K,V> p;
  53.                             binCount = 2;
  54.                             if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
  55.                                                            value)) != null) {
  56.                                 oldVal = p.val;
  57.                                 if (!onlyIfAbsent)
  58.                                     p.val = value;
  59.                             }
  60.                         }
  61.                     }
  62.                 }
  63.                 //链表长度>=8,成立则链表-->树
  64.                 if (binCount != 0) {
  65.                     if (binCount >= TREEIFY_THRESHOLD)
  66.                         treeifyBin(tab, i);
  67.                     if (oldVal != null)
  68.                         return oldVal;
  69.                     break;
  70.                 }
  71.             }
  72.         }
  73.         addCount(1L, binCount);
  74.         return null;
  75.     }

协助扩容helpTransfer

  1.     final 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.             // 根据 length 得到一个标识符号
  6.             int rs = resizeStamp(tab.length);
  7.             while (nextTab == nextTable && table == tab &&
  8.                    (sc = sizeCtl) < 0) {////满足条件，则表示在扩容
  10.                 //判断是否标志发生了变化||  扩容结束了
  11.                 if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
  12.                     //达到最大的扩容参与线程数 ||  判断扩容转移下标是否在调整（扩容结束）
  13.                     sc == rs + MAX_RESIZERS || transferIndex <= 0)
  14.                     break;
  16.                 //执行到这里，说明还在扩容中。并且当前时间可参与扩容
  17.                 //sizectl变量记录参与扩容的线程数。（+1）
  18.                 if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
  19.                     transfer(tab, nextTab);
  20.                     break;
  21.                 }
  22.             }
  23.             return nextTab;
  24.         }
  25.         return table;
  26.     }
  29. 1.检查是否扩容中
  30. 2.sizeCtrl = sizeCtrl+1, 调用transfer()进行真正的扩容。

扩容transfer

  1. 整体思路：
  2. 新建一个nextTab, size是之前的2倍, 将table上的非空元素迁移到nextTab上面去    
  3. private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
  4.         int n = tab.length, stride;
  5.         if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
  6.             //每个线程最少迁移16个槽位
  7.             stride = MIN_TRANSFER_STRIDE; // subdivide range
  8.         if (nextTab == null) {            // initiating
  9.             try {
  10.                 //初始化新的nextTab,容量是现在的2（n << 1）倍
  11.                 Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
  12.                 nextTab = nt;
  13.             } catch (Throwable ex) {      // try to cope with OOME
  14.                 sizeCtl = Integer.MAX_VALUE;
  15.                 return;
  16.             }
  17.             nextTable = nextTab;
  18.             //转移从下标n开始
  19.             transferIndex = n;
  20.         }
  21.         int nextn = nextTab.length;
  22.         ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
  24.         //标记，是否可以向前推进到下一个周期
  25.         boolean advance = true;
  27.         //完成状态，如果是，则结束此方法
  28.         boolean finishing = false; // to ensure sweep before committing nextTab
  29.         for (int i = 0, bound = 0;;) {
  30.             Node<K,V> f; int fh;
  32.             //处理一个周期
  33.             while (advance) {
  34.                 int nextIndex, nextBound;
  36.                 //本线程处理的区间范围为[bound, i),范围还没有处理完成，那么就继续处理
  37.                 if (--i >= bound || finishing)
  38.                     advance = false;
  40.                 //开始找新的一轮的区间
  41.                 else if ((nextIndex = transferIndex) <= 0) {
  42.                     i = -1;
  43.                     advance = false;
  44.                 }
  45.                 //改变transferIndex的值，从16变成了1
  46.                 else if (U.compareAndSwapInt
  47.                          (this, TRANSFERINDEX, nextIndex,
  48.                           nextBound = (nextIndex > stride ?
  49.                                        nextIndex - stride : 0))) {
  50.                     //一块区间最小桶的下标
  51.                     bound = nextBound;
  52.                     //能够处理的最大桶的下标
  53.                     i = nextIndex - 1;
  54.                     advance = false;
  55.                 }
  56.             }
  58.              //每个迁移线程都能达到这里
  59.             if (i < 0 || i >= n || i + n >= nextn) {
  60.                 int sc;
  61.                 if (finishing) {
  62.                     nextTable = null;
  63.                     //table指向临时数组，直接把以前的table丢弃，上面的MOVE等标志全部丢弃，使用新的
  64.                     table = nextTab;
  66.                     //扩大2n-0.5n = 1.50n, 更新新的容量阈值
  67.                     sizeCtl = (n << 1) - (n >>> 1);
  68.                     return;
  69.                 }
  71.                 //当前线程迁移完成
  72.                 if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
  73.                     //注意此时sc的值并不等于sizeCtl，上一步，sizeCtl=sizeCtl-1。这两个对象还是分割的
  74.                     if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
  75.                         return;
  76.                     finishing = advance = true;
  77.                     i = n; // recheck before commit
  78.                 }
  79.             }
  80.             //如果对应槽位为null， 则将ForwardingNode放在对应的地方
  81.             else if ((f = tabAt(tab, i)) == null)
  82.                 advance = casTabAt(tab, i, null, fwd);
  84.             //别的线程已经在处理了，再推进一个下标
  85.             else if ((fh = f.hash) == MOVED)
  86.                 advance = true; // already processed
  87.             else {
  88.             //需要加锁，防止再向里面放值，在放数据时，也会锁住。比如整个table正在迁移，还没有迁移到这个元素，另外一个线程向这个节点插入数据，此时迁移到这里了，会被阻塞住
  89.                 synchronized (f) {
  91.                    //判断i下标和f元素是否相同
  92.                     if (tabAt(tab, i) == f) {
  93.                         //高低位指针
  94.                         Node<K,V> ln, hn;
  95.                         if (fh >= 0) {
  96.                             //n为2^n, 取余后只能是2^n
  97.                             int runBit = fh & n;
  98.                             Node<K,V> lastRun = f;
  99.                             //找到最后一个和fn不相同的节点
  100.                             for (Node<K,V> p = f.next; p != null; p = p.next) {
  101.                                 int b = p.hash & n;
  102.                                 //执行到此处，之后的取值都是一样的，下次循环时，就不用再循环后面的
  103.                                 if (b != runBit) {
  104.                                     runBit = b;
  105.                                     lastRun = p;
  106.                                 }
  107.                             }
  108.                             if (runBit == 0) {
  109.                                 ln = lastRun;
  110.                                 hn = null;
  111.                             }
  112.                             else {
  113.                                 hn = lastRun;
  114.                                 ln = null;
  115.                             }
  116.                             for (Node<K,V> p = f; p != lastRun; p = p.next) {
  117.                                 int ph = p.hash; K pk = p.key; V pv = p.val;
  118.                                 if ((ph & n) == 0)
  119.                                     ln = new Node<K,V>(ph, pk, pv, ln);
  120.                                 else
  121.                                     hn = new Node<K,V>(ph, pk, pv, hn);
  122.                             }
  123.                             setTabAt(nextTab, i, ln);
  124.                             setTabAt(nextTab, i + n, hn);
  125.                             setTabAt(tab, i, fwd);
  126.                             advance = true;
  127.                         }
  128.                         // 红黑树处理逻辑
  129.                         else if (f instanceof TreeBin) {
  130.                             TreeBin<K,V> t = (TreeBin<K,V>)f;
  132.                             //高低位指针
  133.                             TreeNode<K,V> lo = null, loTail = null;
  134.                             TreeNode<K,V> hi = null, hiTail = null;
  135.                             int lc = 0, hc = 0;
  137.                             //中序遍历红黑树
  138.                             for (Node<K,V> e = t.first; e != null; e = e.next) {
  139.                                 int h = e.hash;
  140.                                 TreeNode<K,V> p = new TreeNode<K,V>
  141.                                     (h, e.key, e.val, null, null);
  142.                                 //低位指针处理
  143.                                 if ((h & n) == 0) {
  144.                                     if ((p.prev = loTail) == null)
  145.                                         lo = p;
  146.                                     else
  147.                                         loTail.next = p;
  148.                                     loTail = p;
  149.                                     ++lc;
  150.                                 }
  151.                                 //高位指针处理
  152.                                 else {
  153.                                     if ((p.prev = hiTail) == null)
  154.                                         hi = p;
  155.                                     else
  156.                                         hiTail.next = p;
  157.                                     hiTail = p;
  158.                                     ++hc;
  159.                                 }
  160.                             }
  161.                             //判断是否需要转化为树
  162.                             ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) :
  163.                                 (hc != 0) ? new TreeBin<K,V>(lo) : t;
  164.                             hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) :
  165.                                 (lc != 0) ? new TreeBin<K,V>(hi) : t;
  166.                             setTabAt(nextTab, i, ln);
  167.                             setTabAt(nextTab, i + n, hn);
  168.                             setTabAt(tab, i, fwd);
  169.                             advance = true;
  170.                         }
  171.                     }
  172.                 }
  173.             }
  174.         }
  175.     }

扩容过程总结

两个变量需要了解下:

  • advance: 表示是否可以向下一个轮元素进行迁移。
  • finishing: table所有元素是否迁移完成。

扩容如下:

  • 确定线程每轮迁移元素的个数stride, 比如进来一个线程, 确定扩容table下标为(a,b]之间元素, 下一个线程扩容(b,c]。这里对b-a或者c-b也是由最小值16限制的。 每个线程最少扩容连续16个table的槽位。而标志当前迁移的下标保存在transferIndex。
  • 检查nextTab是否完成初始化, 若没有的话, 说明是第一个迁移的线程, 先初始化nextTab, size是之前table的2倍。
  • 进入while循环查找本轮迁移的table下标元素区间, 保存在(bound, i]中, 注意这里是半开半闭区间。
  • 从i -> bound开始遍历table中每个元素, 这里是从大到小遍历的:
  1. 若该元素为空, 则向该元素标写入ForwardingNode, 然后检查下一个元素。 当别的线程向这个元素插入数据时, 根据这个标志符知道table正在被别的线程迁移, 在putVal中就会调用helpTransfer帮着迁移。
  2. 若该元素的hash=MOVED, 代表次table正在处于迁移之中, 跳过。 按道理不会跑着这里的。
  3. 否则说明该元素跟着的是一个链表或者是个红黑树结构, 若hash>0, 则说明是个链表, 若f instanceof TreeBin, 则说明是个红黑树结构。
  • 链表迁移原理如下: 遍历链表每个节点。 若节点的f.hash&n==0成立, 则将节点放在i, 否则, 则将节点放在n+i上面。

迁移前, 对该元素进行加锁。 遍历链表时, 这里使用lastRun变量, 保留的是上次hash的值, 假如整个链表全部节点f.hash&n==0, 那么第二次遍历, 只要找到lastRun的值, 那么认为之后的节点都是相同值, 减少了不必要的f.hash&n取值。遍历完所有的节点后, 此时形成了两条链表, ln存放的是f.hash&n=0的节点, hn存放的是非0的节点, 然后将ln存放在nextTable第i元素的位置, n+i存放在n+i的位置。
蓝色节点代表:f.hash&n==0, 绿色节点代表f.hash&n!=0。 最终蓝色的节点仍在存放在(0, n)范围里, 绿的节点存放在(n, 2n-1)的范围之内。

  • 迁移链表和红黑树的原理是一样的, 在红黑树中, 我们记录了每个红黑树的first(这个节点不是hash最小的节点)和每个节点的next, 根据这两个元素, 我们可以访问红黑树所有的元素, 红黑树此时也是一个链表, 红黑树和链表迁移的过程一样。红黑树根据迁移后拆分成了hn和ln, 根据链表长度确定链表是红黑树结构还是退化为了链表。

4.如何确定table所有元素迁移完成:

1.7版本了解

分段锁。锁定一级hash表的槽位，意味着锁住整个二级hash表，
一级hash表（segement数组，segement继承了ReentrantLock）、二级hash表
存放数据，需要进行两次hash，并发情况下，第一次hash操作为了获取锁，获取到锁的线程，进行二次hash进行数据存放动作。
image.png