CHM的使用

基本使用同HashMap,

put()
get()
jdk1.8还支持使用lambda表达式
computeIfAbsent : key如果不存在,调用后面的mappingFunction方法
computeIfPresent : key如果存在,调用后面的mappingFunction方法更新值
compute (computeIfPresent 和computeIfAbsent 的结合)
merge :合并数据

CHM的存储结构和实现

1.8去掉了segment,并且引入了红黑树(数组长度大于64，链表长度大于8，则会将链表转为红黑树)，引入红黑树是为了提高检索性能，时间复杂度 O(n)->O(logn) 在这里插入图片描述

CHM源码

put源码分析

   final V putVal(K key, V value, boolean onlyIfAbsent) {if (key == null || value == null) throw new NullPointerException();//计算Hash值int hash = spread(key.hashCode());int binCount = 0;//自旋(;;) CASfor (ConcurrentHashMap.Node<K,V>[] tab = table;;) {ConcurrentHashMap.Node<K,V> f; int n, i, fh;//如果tab为空,则初始化table(需要加锁保证线程安全CAS)if (tab == null || (n = tab.length) == 0)tab = initTable();//初始化完成后进入下一次循环//(n - 1) & hash 计算数组下标， 从内存的偏移量获取值else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {//如果当前node的位置为null,则直接存储到该位置，通过CAS保证原子性if (casTabAt(tab, i, null,new ConcurrentHashMap.Node<K,V>(hash, key, value, null)))break;                   // no lock when adding to empty bin}//如果当前节点状态为MOVED,说明当前该位置处于迁移中,则该线程先帮忙扩容else if ((fh = f.hash) == MOVED)tab = helpTransfer(tab, f);else {V oldVal = null;//锁住当前的node节点 synchronized 保证线程安全synchronized (f) {//重新判断if (tabAt(tab, i) == f) {//针对链表处理, binCount统计当前节点的链表长度if (fh >= 0) {binCount = 1;for (ConcurrentHashMap.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;}ConcurrentHashMap.Node<K,V> pred = e;//如果不存在则将key/value添加到链表中if ((e = e.next) == null) {pred.next = new ConcurrentHashMap.Node<K,V>(hash, key,value, null);break;}}}//针对红黑树else if (f instanceof ConcurrentHashMap.TreeBin) {ConcurrentHashMap.Node<K,V> p;binCount = 2;if ((p = ((ConcurrentHashMap.TreeBin<K,V>)f).putTreeVal(hash, key,value)) != null) {oldVal = p.val;if (!onlyIfAbsent)p.val = value;}}}}if (binCount != 0) {//如果链表长度>=8,会调用treeifyBin方法if (binCount >= TREEIFY_THRESHOLD)//根据阈值和数组大小判断是扩容还是转换为红黑树treeifyBin(tab, i);if (oldVal != null)return oldVal;break;}}}//size++的实现,增加计数addCount(1L, binCount);return null;}

initTable 初始化table

 private final Node<K,V>[] initTable() {Node<K,V>[] tab; int sc;//自旋,只要table没有初始化成功则不断自旋直到完成初始化while ((tab = table) == null || tab.length == 0) {if ((sc = sizeCtl) < 0)Thread.yield(); // lost initialization race; just spin//通过CAS自旋来抢占一个锁标记(sizeCtl),将sizeCtl设置为-1else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {//说明当前线程抢占到了锁try {if ((tab = table) == null || tab.length == 0) {int n = (sc > 0) ? sc : DEFAULT_CAPACITY;@SuppressWarnings("unchecked")Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];table = tab = nt;//sc设置为扩容的阈值(0.75*n)sc = n - (n >>> 2);}} finally {//修改sizeCtl,释放锁sizeCtl = sc;}break;}}return tab;}

sizeCtl不同值表示不同的状态:
在这里插入图片描述

treeifyBin()和tryPresize()

    private final void treeifyBin(ConcurrentHashMap.Node<K,V>[] tab, int index) {ConcurrentHashMap.Node<K,V> b; int n, sc;if (tab != null) {//如果tab长度小于64,调用tryPresize扩容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) {ConcurrentHashMap.TreeNode<K,V> hd = null, tl = null;for (ConcurrentHashMap.Node<K,V> e = b; e != null; e = e.next) {ConcurrentHashMap.TreeNode<K,V> p =new ConcurrentHashMap.TreeNode<K,V>(e.hash, e.key, e.val,null, null);if ((p.prev = tl) == null)hd = p;elsetl.next = p;tl = p;}setTabAt(tab, index, new ConcurrentHashMap.TreeBin<K,V>(hd));}}}}}

创建一个新的数组
对老的数组的数据进行迁移
多线程辅助扩容（针对老的数据，通过多个线程并行来执行数据的迁移过程）
- 记录当前的线程数量（sizeCtl）
- 当每个线程完成数据迁移之后，退出的时候，要减掉协助扩容的线程数量
resizeStamp -> 扩容戳

 //扩容方法private final void tryPresize(int size) {//确定扩容后的目标大小  tableSizeFor会将size转为2的n次方-1的值int c = (size >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY :tableSizeFor(size + (size >>> 1) + 1);int sc;//说明不处于初始化中或者扩容中while ((sc = sizeCtl) >= 0) {ConcurrentHashMap.Node<K,V>[] tab = table; int n;//tab==null则初始化tabif (tab == null || (n = tab.length) == 0) {//实际初始化的大小从sc和c中选择较大的那个n = (sc > c) ? sc : c;//CAS抢占锁标记,将sizeCtl设置为-1if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {try {if (table == tab) {@SuppressWarnings("unchecked")ConcurrentHashMap.Node<K,V>[] nt = (ConcurrentHashMap.Node<K,V>[])new ConcurrentHashMap.Node<?,?>[n];table = nt;sc = n - (n >>> 2);}} finally {sizeCtl = sc;}}}//已经是最大容量了，无法再继续扩容,直接返回else if (c <= sc || n >= MAXIMUM_CAPACITY)break;else if (tab == table) {//多线程参与扩容，需要有一个地方记录有多少个线程参与了扩容(数据迁移),这样才可以判断是否所有线程都已经完成迁移工作int rs = resizeStamp(n);//rs为扩容戳，记录有多少个线程参与扩容，保证当前扩容范围的唯一性//sc<0表示扩容中,第一次进入的时候sc>=0if (sc < 0) {ConcurrentHashMap.Node<K,V>[] nt;//表示扩容结束if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||transferIndex <= 0)break;//表示扩容没有结束,每次进入这个方法则表示有一个新的线程执行transfer方法进行扩容//每增加一个扩容线程,则在低位+1if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))transfer(tab, nt);}//第一次进入的时候,会调用transfer方法进行扩容 CAS成功后,sc会变为负数else if (U.compareAndSwapInt(this, SIZECTL, sc,(rs << RESIZE_STAMP_SHIFT) + 2))transfer(tab, null);}}}

transfer 扩容和数据迁移

数据迁移
- 需要计算当前线程的数据迁移的空间(拆分迁移任务给多个线程)
- 创建一个新的数组(n=old length*2)
- 实现数据迁移(高低位迁移)
  - 如果是红黑树
    数据迁移后，不满足红黑树的条件，则转为链表
  - 如果是链表

在这里插入图片描述

 private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {int n = tab.length, stride;// NCPU 表示当前机器的 CPU 核心数，计算每个线程处理的数据的区间大小stride，最小是16if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)stride = MIN_TRANSFER_STRIDE; // subdivide range//表示扩容之后的数组，在原来的基础上扩大两倍。if (nextTab == null) { // initiatingtry {@SuppressWarnings("unchecked")Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];nextTab = nt;} catch (Throwable ex) { // try to cope with OOMEsizeCtl = Integer.MAX_VALUE;return;}nextTable = nextTab;//transferIndex = old table[] 的长度。transferIndex = n;}int nextn = nextTab.length;//用来表示已经迁移完的状态，也就是说，如果某个old数组的节点完成了迁移，则需要更改成fwd。(fwd的hash为-1,就是MOVED值)ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);boolean advance = true;boolean finishing = false; // to ensure sweep before committing nextTab//死循环,完成扩容后代码里会有退出循环的条件 finishing==true则结束循环for (int i = 0, bound = 0;;) {Node<K,V> f; int fh;//计算要迁移的数组区间while (advance) {int nextIndex, nextBound;//--i>= bound,说明当前数据还是在本次的处理范围内,那么不需要重新计算区间if (--i >= bound || finishing)advance = false;else if ((nextIndex = transferIndex) <= 0) {i = -1;advance = false;}//stride为每个线程要处理的区间大小,transferIndex初始值为oldTab的大小，这里通过CAS//更新transferIndex的值else if (U.compareAndSwapInt(this, TRANSFERINDEX, nextIndex,nextBound = (nextIndex > stride ?nextIndex - stride : 0))) {bound = nextBound;//计算需要处理的区间后，当前线程要处理的数据范围就是[nextBound,i)i = nextIndex - 1;advance = false;}//假设数组长度是32,//第一次 [16(nextBound),31(i)]//第二次 [0,15]}//判断是否扩容结束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)) {//对应于前面的+2if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)return;finishing = advance = true;i = n; // recheck before commit}}//第一次i=31,则得到数组下标为31的位置的值。(这里就是old tab下标最大的那个位置的值)else if ((f = tabAt(tab, i)) == null) //说明当前数组位置为空,不需要迁移。advance = casTabAt(tab, i, null, fwd); //直接改成fwd -> 表示迁移完成// 判断是否已经被处理过了，如果是则进入下一次区间遍历(MOVED是fwd节点的初始状态)// 下一次遍历因为--i>= bound所以又会走到这里的逻辑else if ((fh = f.hash) == MOVED)advance = true; // already processedelse {//加锁->针对当前要去迁移的节点,保证迁移过程中，其他线程调用put()向这个位置插入数据时,必须要等待。synchronized (f) { ////针对不同类型的节点，做不同的处理 链表 or 红黑树//链表迁移的时候,由于数组长度改变,那么同一个key计算出来的数组下标也不一样了,需要重新计算if (tabAt(tab, i) == f) {//两个链表,低位链表,高位链表, 低位链表示不需要更改数组下标的链表(1)高低链表单独讲解下Node<K,V> ln, hn;//fh表示当前节点的hashcodeif (fh >= 0) {int runBit = fh & n;Node<K,V> lastRun = f;for (Node<K,V> p = f.next; p != null; p = p.next) {int b = p.hash & n;if (b != runBit) {runBit = b;lastRun = p;}}//runBit==0表示不需要更改数组下标if (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);elsehn = new Node<K,V>(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<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;elseloTail.next = p;loTail = p;++lc;}else {if ((p.prev = hiTail) == null)hi = p;elsehiTail.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;}}}}}}

高低位的迁移

先来讲一下为什么可以做高低位的迁移呢?

之前源码里数组下标的计算方式是： i = (n - 1) & hash
在扩容的时候n的值是变化的,会变为2*n，不同的key的hash值不同，所以扩容后，部分key可能会发生迁移(计算出来的下标不一样了)

注意，我们这里的n是2的整数次幂,那么n-1的二进制表示就会从1xx111 (原来有k个1),变为11xx111(k+1个1), 和hash进行与运算之后,其结果和之前的相比，可能会在第k+1位多了一个1，其他位和旧tab下的运算结果是一致的

转成十进制其实就是+n,是否下标会从i->n+i取决于key的hash的k+1位是1还是0(从低到高的k+1位)

int runBit = fh & n;就是用来判断原来的hash的第k+1位是否为1,如果runBit==0，则说明不是1(说明数据不会发生迁移)，否则就是1

    //fh表示当前节点的hashcode,如果fh<0则表示该节点已经迁移完成了if (fh >= 0) {//runBit==0说明数据不会发生迁移int runBit = fh & n;Node<K,V> lastRun = f;//lastRun 机制选出最后hash值相同的链表的头节点(lastRun节点之后的节点的迁移后都在一个位置上)for (Node<K,V> p = f.next; p != null; p = p.next) {int b = p.hash & n;//该节点和前一个节点的runBit不一样,则会进入下面的代码if (b != runBit) {runBit = b;lastRun = p;}}//根据最高位Bit为1还是0，决定放入新数组的高位还是低位if (runBit == 0) {ln = lastRun;hn = null;}else {hn = lastRun;ln = null;}//迁移剩余的节点，ph & n == 0存入低位，1则存入高位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);elsehn = new Node<K,V>(ph, pk, pv, hn);}//将低位链表放在i的位置上setTabAt(nextTab, i, ln);//将高位链表放在i的位置上setTabAt(nextTab, i + n, hn);//更新fwd标识,表示已经迁移完数据setTabAt(tab, i, fwd);advance = true;}