一、Unsafe介绍

1、Unsafe简介

Unsafe类相当于是一个java语言中的后门类，提供了硬件级别的原子操作，所以在一些并发编程中被大量使用。jdk已经作出说明，该类对程序员而言不是一个安全操作，在后续的jdk升级过程中，可能会禁用该类。所以这个类的使用是一把双刃剑，实际项目中谨慎使用，以免造成jdk升级不兼容问题。

2、Unsafe Api

这里并不系统讲解Unsafe的所有功能，只介绍和接下来内容相关的操作
arrayBaseOffset：获取数组的基础偏移量
arrayIndexScale：获取数组中元素的偏移间隔，要获取对应所以的元素，将索引号和该值相乘，获得数组中指定角标元素的偏移量
getObjectVolatile：获取对象上的属性值或者数组中的元素
getObject：获取对象上的属性值或者数组中的元素，已过时
putOrderedObject：设置对象的属性值或者数组中某个角标的元素，不保证线程间即时可见性，更高效
putObjectVolatile：设置对象的属性值或者数组中某个角标的元素
putObject：设置对象的属性值或者数组中某个角标的元素，已过时

3、代码演示

  1. public class Test02 {
  3.     public static void main(String[] args) throws Exception {
  4.         Integer[] arr = {2,5,1,8,10};
  6.         //获取Unsafe对象
  7.         Unsafe unsafe = getUnsafe();
  9.         //获取Integer[]的基础偏移量
  10.         int baseOffset = unsafe.arrayBaseOffset(Integer[].class);
  12.         //获取Integer[]中元素的偏移间隔
  13.         int indexScale = unsafe.arrayIndexScale(Integer[].class);
  15.         //获取数组中索引为2的元素对象
  16.         Object o = unsafe.getObjectVolatile(arr, (2 * indexScale) + baseOffset);
  17.         System.out.println(o); //1
  19.         //设置数组中索引为2的元素值为100
  20.         unsafe.putOrderedObject(arr,(2 * indexScale) + baseOffset,100);
  22.         System.out.println(Arrays.toString(arr));//[2, 5, 100, 8, 10]
  23.     }
  25.     //反射获取Unsafe对象
  26.     public static Unsafe getUnsafe() throws Exception {
  27.         Field theUnsafe = Unsafe.class.getDeclaredField("theUnsafe");
  28.         theUnsafe.setAccessible(true);
  29.         return (Unsafe) theUnsafe.get(null);
  30.     }
  31. }

3.1 图解说明

微信截图_20200417093442.png

二、jdk1.7容器初始化

1、源码解析

  1. // 空参构造
  2. public ConcurrentHashMap() {
  4.     //调用本类的带参构造
  5.     //DEFAULT_INITIAL_CAPACITY = 16
  6.     //DEFAULT_LOAD_FACTOR = 0.75f
  7.     //int DEFAULT_CONCURRENCY_LEVEL = 16
  8.     this(DEFAULT_INITIAL_CAPACITY, DEFAULT_LOAD_FACTOR, DEFAULT_CONCURRENCY_LEVEL);
  9. }

三个参数的构造：一些非核心逻辑的代码已经省略

  1. //initialCapacity 定义ConcurrentHashMap存放元素的容量
  2. //concurrencyLevel 定义ConcurrentHashMap中Segment[]的大小
  3. public ConcurrentHashMap(int initialCapacity,
  4.                          float loadFactor, int concurrencyLevel) {
  6.     int sshift = 0;
  7.     int ssize = 1;
  8.     //计算Segment[]的大小，保证是2的幂次方数
  9.     while (ssize < concurrencyLevel) {
  10.         ++sshift;
  11.         ssize <<= 1;
  12.     }
  13.     //这两个值用于后面计算Segment[]的角标
  14.     this.segmentShift = 32 - sshift;    // 28
  15.     this.segmentMask = ssize - 1;        // 15
  17.     //计算每个Segment中存储元素的个数
  18.     int c = initialCapacity / ssize;    // 16 / 16 = 1
  19.     if (c * ssize < initialCapacity)
  20.         ++c;
  21.     //最小Segment中存储元素的个数为2
  22.     int cap = MIN_SEGMENT_TABLE_CAPACITY;
  23.     ////矫正每个Segment中存储元素的个数，保证是2的幂次方，最小为2
  24.     while (cap < c)
  25.         cap <<= 1;
  26.     //创建一个Segment对象，作为其他Segment对象的模板
  27.     Segment<K,V> s0 =
  28.         new Segment<K,V>(loadFactor, (int)(cap * loadFactor),
  29.                          (HashEntry<K,V>[])new HashEntry[cap]);
  30.     Segment<K,V>[] ss = (Segment<K,V>[])new Segment[ssize];
  31.     //利用Unsafe类，将创建的Segment对象存入0角标位置
  32.     UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]
  33.     this.segments = ss;
  34. }

综上：ConcurrentHashMap中保存了一个默认长度为16的Segment[]，每个Segment元素中保存了一个默认长度为2的HashEntry[]，我们添加的元素，是存入对应的Segment中的HashEntry[]中。所以ConcurrentHashMap中默认元素的长度是32个，而不是16个

2、图解

微信截图_20200417101208.png

3、Segment数组

  1. static final class Segment<K,V> extends ReentrantLock implements Serializable {
  2.     ...
  3. }

我们发现Segment是继承自ReentrantLock的，学过线程的兄弟都知道，它可以实现同步操作，从而保证多线程下的安全。因为每个Segment之间的锁互不影响，所以我们也将ConcurrentHashMap中的这种锁机制称之为分段锁，这比HashTable的线程安全操作高效的多。

4、HashEntry数组

  1. //ConcurrentHashMap中真正存储数据的对象
  2. static final class HashEntry<K,V> {
  3.     final int hash; //通过运算，得到的键的hash值
  4.     final K key; // 存入的键
  5.     volatile V value; //存入的值
  6.     volatile HashEntry<K,V> next; //记录下一个元素，形成单向链表
  8.     HashEntry(int hash, K key, V value, HashEntry<K,V> next) {
  9.         this.hash = hash;
  10.         this.key = key;
  11.         this.value = value;
  12.         this.next = next;
  13.     }
  14. }

image.png

三、jdk1.7添加安全

1、源码分析

1.1 ConcurrentHashMap的put方法

  1. public V put(K key, V value) {
  2.     Segment<K,V> s;
  3.     if (value == null)
  4.         throw new NullPointerException();
  5.     //基于key，计算hash值
  6.     int hash = hash(key);
  7.     //因为一个键要计算两个数组的索引，为了避免冲突，这里取高位计算Segment[]的索引
  8.     int j = (hash >>> segmentShift) & segmentMask;
  9.     //判断该索引位的Segment对象是否创建，没有就创建
  10.     if ((s = (Segment<K,V>)UNSAFE.getObject          // nonvolatile; recheck
  11.          (segments, (j << SSHIFT) + SBASE)) == null) //  in ensureSegment
  12.         s = ensureSegment(j);
  13.     //调用Segmetn的put方法实现元素添加
  14.     return s.put(key, hash, value, false);
  15. }

1.2 ConcurrentHashMap的ensureSegment方法

  1. //创建对应索引位的Segment对象，并返回
  2. private Segment<K,V> ensureSegment(int k) {
  3.     final Segment<K,V>[] ss = this.segments;
  4.     long u = (k << SSHIFT) + SBASE; // raw offset
  5.     Segment<K,V> seg;
  6.     //获取，如果为null，即创建
  7.     if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u)) == null) {
  8.         //以0角标位的Segment为模板
  9.         Segment<K,V> proto = ss[0]; // use segment 0 as prototype
  10.         int cap = proto.table.length;
  11.         float lf = proto.loadFactor;
  12.         int threshold = (int)(cap * lf);
  13.         HashEntry<K,V>[] tab = (HashEntry<K,V>[])new HashEntry[cap];
  14.         //获取，如果为null，即创建
  15.         if ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
  16.             == null) { // recheck
  17.             //创建
  18.             Segment<K,V> s = new Segment<K,V>(lf, threshold, tab);
  19.             //自旋方式，将创建的Segment对象放到Segment[]中，确保线程安全
  20.             while ((seg = (Segment<K,V>)UNSAFE.getObjectVolatile(ss, u))
  21.                    == null) {
  22.                 if (UNSAFE.compareAndSwapObject(ss, u, null, seg = s))
  23.                     break;
  24.             }
  25.         }
  26.     }
  27.     //返回
  28.     return seg;
  29. }

1.3 Segment的put方法

  1. final V put(K key, int hash, V value, boolean onlyIfAbsent) {
  2.     //尝试获取锁，获取成功，node为null，代码向下执行
  3.     //如果有其他线程占据锁对象，那么去做别的事情，而不是一直等待，提升效率
  4.     //scanAndLockForPut 稍后分析
  5.     HashEntry<K,V> node = tryLock() ? null :
  6.         scanAndLockForPut(key, hash, value);
  7.     V oldValue;
  8.     try {
  9.         HashEntry<K,V>[] tab = table;
  10.         //取hash的低位，计算HashEntry[]的索引
  11.         int index = (tab.length - 1) & hash;
  12.         //获取索引位的元素对象
  13.         HashEntry<K,V> first = entryAt(tab, index);
  14.         for (HashEntry<K,V> e = first;;) {
  15.             //获取的元素对象不为空
  16.             if (e != null) {
  17.                 K k;
  18.                 //如果是重复元素，覆盖原值
  19.                 if ((k = e.key) == key ||
  20.                     (e.hash == hash && key.equals(k))) {
  21.                     oldValue = e.value;
  22.                     if (!onlyIfAbsent) {
  23.                         e.value = value;
  24.                         ++modCount;
  25.                     }
  26.                     break;
  27.                 }
  28.                 //如果不是重复元素，获取链表的下一个元素，继续循环遍历链表
  29.                 e = e.next;
  30.             }
  31.             else { //如果获取到的元素为空
  32.                 //当前添加的键值对的HashEntry对象已经创建
  33.                 if (node != null)
  34.                     node.setNext(first); //头插法关联即可
  35.                 else
  36.                     //创建当前添加的键值对的HashEntry对象
  37.                     node = new HashEntry<K,V>(hash, key, value, first);
  38.                 //添加的元素数量递增
  39.                 int c = count + 1;
  40.                 //判断是否需要扩容
  41.                 if (c > threshold && tab.length < MAXIMUM_CAPACITY)
  42.                     //需要扩容
  43.                     rehash(node);
  44.                 else
  45.                     //不需要扩容
  46.                     //将当前添加的元素对象，存入数组角标位，完成头插法添加元素
  47.                     setEntryAt(tab, index, node);
  48.                 ++modCount;
  49.                 count = c;
  50.                 oldValue = null;
  51.                 break;
  52.             }
  53.         }
  54.     } finally {
  55.         //释放锁
  56.         unlock();
  57.     }
  58.     return oldValue;
  59. }

1.4 Segment的scanAndLockForPut方法

该方法在线程没有获取到锁的情况下，去完成HashEntry对象的创建，提升效率
但是这个操作个人感觉有点累赘了。

  1. private HashEntry<K,V> scanAndLockForPut(K key, int hash, V value) {
  2.     //获取头部元素
  3.     HashEntry<K,V> first = entryForHash(this, hash);
  4.     HashEntry<K,V> e = first;
  5.     HashEntry<K,V> node = null；
  6.     int retries = -1; // negative while locating node
  7.     while (!tryLock()) {
  8.         //获取锁失败
  9.         HashEntry<K,V> f; // to recheck first below
  10.         if (retries < 0) {
  11.             //没有下一个节点，并且也不是重复元素，创建HashEntry对象，不再遍历
  12.             if (e == null) {
  13.                 if (node == null) // speculatively create node
  14.                     node = new HashEntry<K,V>(hash, key, value, null);
  15.                 retries = 0;
  16.             }
  17.             else if (key.equals(e.key))
  18.                 //重复元素，不创建HashEntry对象，不再遍历
  19.                 retries = 0;
  20.             else
  21.                 //继续遍历下一个节点
  22.                 e = e.next;
  23.         }
  24.         else if (++retries > MAX_SCAN_RETRIES) {
  25.             //如果尝试获取锁的次数过多，直接阻塞
  26.             //MAX_SCAN_RETRIES会根据可用cpu核数来确定
  27.             lock();
  28.             break;
  29.         }
  30.         else if ((retries & 1) == 0 &&
  31.                  (f = entryForHash(this, hash)) != first) {
  32.             //如果期间有别的线程获取锁，重新遍历
  33.             e = first = f; // re-traverse if entry changed
  34.             retries = -1;
  35.         }
  36.     }
  37.     return node;
  38. }

2、模拟多线程的代码流程

这里“通话”和“重地”的哈希值是一样的，那么他们添加时，会存入同一个Segment对象，必然会存在锁竞争

  1. public static void main(String[] args) throws Exception {
  2.     final ConcurrentHashMap chm = new ConcurrentHashMap();
  4.     new Thread(){
  5.         @Override
  6.         public void run() {
  7.             chm.put("通话","11");
  8.             System.out.println("-----------");
  9.         }
  10.     }.start();
  12.     //让第一个线程先启动，进入put方法
  13.     Thread.sleep(1000);
  15.     new Thread(){
  16.         @Override
  17.         public void run() {
  18.             chm.put("重地","22");
  19.             System.out.println("===========");
  20.         }
  21.     }.start();
  22. }

2.1 多线程环境下的条件断点设置

微信图片_20200417111509.png

2.2 运行结果

会发现两个线程，分别停在不同的断点位置，这就是多线程锁互斥产生的结果
然后就可以分别让不同的线程向下执行，查看代码走向了。
微信截图_20200417111834.png

四、jdk1.7扩容安全

1、源码分析

  1. private void rehash(HashEntry<K,V> node) {
  2.     HashEntry<K,V>[] oldTable = table;
  3.     int oldCapacity = oldTable.length;
  4.     //两倍容量
  5.     int newCapacity = oldCapacity << 1;
  6.     threshold = (int)(newCapacity * loadFactor);
  7.     //基于新容量，创建HashEntry数组
  8.     HashEntry<K,V>[] newTable =
  9.         (HashEntry<K,V>[]) new HashEntry[newCapacity];
  10.     int sizeMask = newCapacity - 1;
  11.        //实现数据迁移
  12.     for (int i = 0; i < oldCapacity ; i++) {
  13.         HashEntry<K,V> e = oldTable[i];
  14.         if (e != null) {
  15.             HashEntry<K,V> next = e.next;
  16.             int idx = e.hash & sizeMask;
  17.             if (next == null)   //  Single node on list
  18.                 //原位置只有一个元素，直接放到新数组即可
  19.                 newTable[idx] = e;
  20.             else { // Reuse consecutive sequence at same slot
  21.                 //=========图一=====================
  22.                 HashEntry<K,V> lastRun = e;
  23.                 int lastIdx = idx;
  24.                 for (HashEntry<K,V> last = next;
  25.                      last != null;
  26.                      last = last.next) {
  27.                     int k = last.hash & sizeMask;
  28.                     if (k != lastIdx) {
  29.                         lastIdx = k;
  30.                         lastRun = last;
  31.                     }
  32.                 }
  33.                 //=========图一=====================
  35.                 //=========图二=====================
  36.                 newTable[lastIdx] = lastRun;
  37.                 //=========图二=====================
  38.                 // Clone remaining nodes
  39.                 //=========图三=====================
  40.                 for (HashEntry<K,V> p = e; p != lastRun; p = p.next) {
  41.                     V v = p.value;
  42.                     int h = p.hash;
  43.                     int k = h & sizeMask;
  44.                     HashEntry<K,V> n = newTable[k];
  45.                     //这里旧的HashEntry不会放到新数组
  46.                     //而是基于原来的数据创建了一个新的HashEntry对象，放入新数组
  47.                     newTable[k] = new HashEntry<K,V>(h, p.key, v, n);
  48.                 }
  49.                 //=========图三=====================
  50.             }
  51.         }
  52.     }
  53.     //采用头插法，将新元素加入到数组中
  54.     int nodeIndex = node.hash & sizeMask; // add the new node
  55.     node.setNext(newTable[nodeIndex]);
  56.     newTable[nodeIndex] = node;
  57.     table = newTable;
  58. }

2、图解

图一
微信截图_20200417113823.png
图二
微信截图_20200417114240.png
图三
微信截图_20200417114954.png

五、jdk1.7集合长度获取

1、源码分析

  1. public int size() {
  2.     // Try a few times to get accurate count. On failure due to
  3.     // continuous async changes in table, resort to locking.
  4.     final Segment<K,V>[] segments = this.segments;
  5.     int size;
  6.     boolean overflow; // true if size overflows 32 bits
  7.     long sum;         // sum of modCounts
  8.     long last = 0L;   // previous sum
  9.     int retries = -1; // first iteration isn't retry
  10.     try {
  11.         for (;;) {
  12.             //当第5次走到这个地方时，会将整个Segment[]的所有Segment对象锁住
  13.             if (retries++ == RETRIES_BEFORE_LOCK) {
  14.                 for (int j = 0; j < segments.length; ++j)
  15.                     ensureSegment(j).lock(); // force creation
  16.             }
  17.             sum = 0L;
  18.             size = 0;
  19.             overflow = false;
  20.             for (int j = 0; j < segments.length; ++j) {
  21.                 Segment<K,V> seg = segmentAt(segments, j);
  22.                 if (seg != null) {
  23.                     //累加所有Segment的操作次数
  24.                     sum += seg.modCount;
  25.                     int c = seg.count;
  26.                     //累加所有segment中的元素个数 size+=c
  27.                     if (c < 0 || (size += c) < 0)
  28.                         overflow = true;
  29.                 }
  30.             }
  31.             //当这次累加值和上一次累加值一样，证明没有进行新的增删改操作，返回sum
  32.             //第一次last为0，如果有元素的话，这个for循环最少循环两次的
  33.             if (sum == last)
  34.                 break;
  35.             //记录累加的值
  36.             last = sum;
  37.         }
  38.     } finally {
  39.         //如果之前有锁住，解锁
  40.         if (retries > RETRIES_BEFORE_LOCK) {
  41.             for (int j = 0; j < segments.length; ++j)
  42.                 segmentAt(segments, j).unlock();
  43.         }
  44.     }
  45.     //溢出，返回int的最大值，否则返回累加的size
  46.     return overflow ? Integer.MAX_VALUE : size;
  47. }

说明：市面上很多讲ConcurrentHashMap的源码分析课程，大多都是以营销为目的，并没有完整讲解添加安全和扩容安全，就目前我所知，所有公开课程中，没有任何课程讲解jdk1.8多线程扩容效率的改进方案

六、jdk1.8容器初始化

1、源码分析

在jdk8的ConcurrentHashMap中一共有5个构造方法，这四个构造方法中都没有对内部的数组做初始化， 只是对一些变量的初始值做了处理
jdk8的ConcurrentHashMap的数组初始化是在第一次添加元素时完成

  1. //没有维护任何变量的操作，如果调用该方法，数组长度默认是16
  2. public ConcurrentHashMap() {
  3. }
  6. //传递进来一个初始容量，ConcurrentHashMap会基于这个值计算一个比这个值大的2的幂次方数作为初始容量
  7. public ConcurrentHashMap(int initialCapacity) {
  8.     if (initialCapacity < 0)
  9.         throw new IllegalArgumentException();
  10.     int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?
  11.                MAXIMUM_CAPACITY :
  12.                tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));
  13.     this.sizeCtl = cap;
  14. }

注意：调用这个方法，得到的初始容量和我们之前讲的HashMap以及jdk7的ConcurrentHashMap不同，即使你传递的是一个2的幂次方数，该方法计算出来的初始容量依然是比这个值大的2的幂次方数

  1. //调用四个参数的构造
  2. public ConcurrentHashMap(int initialCapacity, float loadFactor) {
  3.     this(initialCapacity, loadFactor, 1);
  4. }
  7. //计算一个大于或者等于给定的容量值，该值是2的幂次方数作为初始容量
  8. public ConcurrentHashMap(int initialCapacity,
  9.                          float loadFactor, int concurrencyLevel) {
  10.     if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0)
  11.         throw new IllegalArgumentException();
  12.     if (initialCapacity < concurrencyLevel)   // Use at least as many bins
  13.         initialCapacity = concurrencyLevel;   // as estimated threads
  14.     long size = (long)(1.0 + (long)initialCapacity / loadFactor);
  15.     int cap = (size >= (long)MAXIMUM_CAPACITY) ?
  16.         MAXIMUM_CAPACITY : tableSizeFor((int)size);
  17.     this.sizeCtl = cap;
  18. }
  21. //基于一个Map集合，构建一个ConcurrentHashMap
  22. //初始容量为16
  23. public ConcurrentHashMap(Map<? extends K, ? extends V> m) {
  24.     this.sizeCtl = DEFAULT_CAPACITY;
  25.     putAll(m);
  26. }

2、sizeCtl含义解释

注意：以上这些构造方法中，都涉及到一个变量sizeCtl，这个变量是一个非常重要的变量，而且具有非常丰富的含义，它的值不同，对应的含义也不一样，这里我们先对这个变量不同的值的含义做一下说明，后续源码分析过程中，进一步解释
sizeCtl为0，代表数组未初始化， 且数组的初始容量为16
sizeCtl为正数，如果数组未初始化，那么其记录的是数组的初始容量，如果数组已经初始化，那么其记录的是数组的扩容阈值
sizeCtl为-1，表示数组正在进行初始化
sizeCtl小于0，并且不是-1，表示数组正在扩容， -(1+n)，表示此时有n个线程正在共同完成数组的扩容操作

七、jdk1.8添加安全

1、源码分析

1.1 添加元素put/putVal方法

  1. public V put(K key, V value) {
  2.     return putVal(key, value, false);
  3. }
  6. final V putVal(K key, V value, boolean onlyIfAbsent) {
  7.     //如果有空值或者空键，直接抛异常
  8.     if (key == null || value == null) throw new NullPointerException();
  9.     //基于key计算hash值，并进行一定的扰动
  10.     int hash = spread(key.hashCode());
  11.     //记录某个桶上元素的个数，如果超过8个，会转成红黑树
  12.     int binCount = 0;
  13.     for (Node<K,V>[] tab = table;;) {
  14.         Node<K,V> f; int n, i, fh;
  15.         //如果数组还未初始化，先对数组进行初始化
  16.         if (tab == null || (n = tab.length) == 0)
  17.             tab = initTable();
  18.         //如果hash计算得到的桶位置没有元素，利用cas将元素添加
  19.         else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
  20.             //cas+自旋（和外侧的for构成自旋循环），保证元素添加安全
  21.             if (casTabAt(tab, i, null,
  22.                          new Node<K,V>(hash, key, value, null)))
  23.                 break;                   // no lock when adding to empty bin
  24.         }
  25.         //如果hash计算得到的桶位置元素的hash值为MOVED，证明正在扩容，那么协助扩容
  26.         else if ((fh = f.hash) == MOVED)
  27.             tab = helpTransfer(tab, f);
  28.         else {
  29.             //hash计算的桶位置元素不为空，且当前没有处于扩容操作，进行元素添加
  30.             V oldVal = null;
  31.             //对当前桶进行加锁，保证线程安全，执行元素添加操作
  32.             synchronized (f) {
  33.                 if (tabAt(tab, i) == f) {
  34.                     //普通链表节点
  35.                     if (fh >= 0) {
  36.                         binCount = 1;
  37.                         for (Node<K,V> e = f;; ++binCount) {
  38.                             K ek;
  39.                             if (e.hash == hash &&
  40.                                 ((ek = e.key) == key ||
  41.                                  (ek != null && key.equals(ek)))) {
  42.                                 oldVal = e.val;
  43.                                 if (!onlyIfAbsent)
  44.                                     e.val = value;
  45.                                 break;
  46.                             }
  47.                             Node<K,V> pred = e;
  48.                             if ((e = e.next) == null) {
  49.                                 pred.next = new Node<K,V>(hash, key,
  50.                                                           value, null);
  51.                                 break;
  52.                             }
  53.                         }
  54.                     }
  55.                     //树节点，将元素添加到红黑树中
  56.                     else if (f instanceof TreeBin) {
  57.                         Node<K,V> p;
  58.                         binCount = 2;
  59.                         if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
  60.                                                        value)) != null) {
  61.                             oldVal = p.val;
  62.                             if (!onlyIfAbsent)
  63.                                 p.val = value;
  64.                         }
  65.                     }
  66.                 }
  67.             }
  68.             if (binCount != 0) {
  69.                 //链表长度大于/等于8，将链表转成红黑树
  70.                 if (binCount >= TREEIFY_THRESHOLD)
  71.                     treeifyBin(tab, i);
  72.                 //如果是重复键，直接将旧值返回
  73.                 if (oldVal != null)
  74.                     return oldVal;
  75.                 break;
  76.             }
  77.         }
  78.     }
  79.     //添加的是新元素，维护集合长度，并判断是否要进行扩容操作
  80.     addCount(1L, binCount);
  81.     return null;
  82. }

通过以上源码，我们可以看到，当需要添加元素时，会针对当前元素所对应的桶位进行加锁操作，这样一方面保证元素添加时，多线程的安全，同时对某个桶位加锁不会影响其他桶位的操作，进一步提升多线程的并发效率

1.2 数组初始化，initTable方法

  1. private final Node<K,V>[] initTable() {
  2.     Node<K,V>[] tab; int sc;
  3.     //cas+自旋，保证线程安全，对数组进行初始化操作
  4.     while ((tab = table) == null || tab.length == 0) {
  5.         //如果sizeCtl的值（-1）小于0，说明此时正在初始化， 让出cpu
  6.         if ((sc = sizeCtl) < 0)
  7.             Thread.yield(); // lost initialization race; just spin
  8.         //cas修改sizeCtl的值为-1，修改成功，进行数组初始化，失败，继续自旋
  9.         else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
  10.             try {
  11.                 if ((tab = table) == null || tab.length == 0) {
  12.                     //sizeCtl为0，取默认长度16，否则去sizeCtl的值
  13.                     int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
  14.                     @SuppressWarnings("unchecked")
  15.                     //基于初始长度，构建数组对象
  16.                     Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
  17.                     table = tab = nt;
  18.                     //计算扩容阈值，并赋值给sc
  19.                     sc = n - (n >>> 2);
  20.                 }
  21.             } finally {
  22.                 //将扩容阈值，赋值给sizeCtl
  23.                 sizeCtl = sc;
  24.             }
  25.             break;
  26.         }
  27.     }
  28.     return tab;
  29. }

2、图解

2.1 put加锁图解

Snipaste_2020-05-16_12-04-08.jpg

八、jdk1.8扩容安全

1、源码分析

  1. private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
  2.     int n = tab.length, stride;
  3.     //如果是多cpu，那么每个线程划分任务，最小任务量是16个桶位的迁移
  4.     if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
  5.         stride = MIN_TRANSFER_STRIDE; // subdivide range
  6.     //如果是扩容线程，此时新数组为null
  7.     if (nextTab == null) {            // initiating
  8.         try {
  9.             @SuppressWarnings("unchecked")
  10.             //两倍扩容创建新数组
  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.         //记录线程开始迁移的桶位，从后往前迁移
  19.         transferIndex = n;
  20.     }
  21.     //记录新数组的末尾
  22.     int nextn = nextTab.length;
  23.     //已经迁移的桶位，会用这个节点占位（这个节点的hash值为-1--MOVED）
  24.     ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
  25.     boolean advance = true;
  26.     boolean finishing = false; // to ensure sweep before committing nextTab
  27.     for (int i = 0, bound = 0;;) {
  28.         Node<K,V> f; int fh;
  29.         while (advance) {
  30.             int nextIndex, nextBound;
  31.             //i记录当前正在迁移桶位的索引值
  32.             //bound记录下一次任务迁移的开始桶位
  34.             //--i >= bound 成立表示当前线程分配的迁移任务还没有完成
  35.             if (--i >= bound || finishing)
  36.                 advance = false;
  37.             //没有元素需要迁移 -- 后续会去将扩容线程数减1，并判断扩容是否完成
  38.             else if ((nextIndex = transferIndex) <= 0) {
  39.                 i = -1;
  40.                 advance = false;
  41.             }
  42.             //计算下一次任务迁移的开始桶位，并将这个值赋值给transferIndex
  43.             else if (U.compareAndSwapInt
  44.                      (this, TRANSFERINDEX, nextIndex,
  45.                       nextBound = (nextIndex > stride ?
  46.                                    nextIndex - stride : 0))) {
  47.                 bound = nextBound;
  48.                 i = nextIndex - 1;
  49.                 advance = false;
  50.             }
  51.         }
  52.         //如果没有更多的需要迁移的桶位，就进入该if
  53.         if (i < 0 || i >= n || i + n >= nextn) {
  54.             int sc;
  55.             //扩容结束后，保存新数组，并重新计算扩容阈值，赋值给sizeCtl
  56.             if (finishing) {
  57.                 nextTable = null;
  58.                 table = nextTab;
  59.                 sizeCtl = (n << 1) - (n >>> 1);
  60.                 return;
  61.             }
  62.            //扩容任务线程数减1
  63.             if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
  64.                 //判断当前所有扩容任务线程是否都执行完成
  65.                 if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
  66.                     return;
  67.                 //所有扩容线程都执行完，标识结束
  68.                 finishing = advance = true;
  69.                 i = n; // recheck before commit
  70.             }
  71.         }
  72.         //当前迁移的桶位没有元素，直接在该位置添加一个fwd节点
  73.         else if ((f = tabAt(tab, i)) == null)
  74.             advance = casTabAt(tab, i, null, fwd);
  75.         //当前节点已经被迁移
  76.         else if ((fh = f.hash) == MOVED)
  77.             advance = true; // already processed
  78.         else {
  79.             //当前节点需要迁移，加锁迁移，保证多线程安全
  80.             //此处迁移逻辑和jdk7的ConcurrentHashMap相同，不再赘述
  81.             synchronized (f) {
  82.                 if (tabAt(tab, i) == f) {
  83.                     Node<K,V> ln, hn;
  84.                     if (fh >= 0) {
  85.                         int runBit = fh & n;
  86.                         Node<K,V> lastRun = f;
  87.                         for (Node<K,V> p = f.next; p != null; p = p.next) {
  88.                             int b = p.hash & n;
  89.                             if (b != runBit) {
  90.                                 runBit = b;
  91.                                 lastRun = p;
  92.                             }
  93.                         }
  94.                         if (runBit == 0) {
  95.                             ln = lastRun;
  96.                             hn = null;
  97.                         }
  98.                         else {
  99.                             hn = lastRun;
  100.                             ln = null;
  101.                         }
  102.                         for (Node<K,V> p = f; p != lastRun; p = p.next) {
  103.                             int ph = p.hash; K pk = p.key; V pv = p.val;
  104.                             if ((ph & n) == 0)
  105.                                 ln = new Node<K,V>(ph, pk, pv, ln);
  106.                             else
  107.                                 hn = new Node<K,V>(ph, pk, pv, hn);
  108.                         }
  109.                         setTabAt(nextTab, i, ln);
  110.                         setTabAt(nextTab, i + n, hn);
  111.                         setTabAt(tab, i, fwd);
  112.                         advance = true;
  113.                     }
  114.                     else if (f instanceof TreeBin) {
  115.                         TreeBin<K,V> t = (TreeBin<K,V>)f;
  116.                         TreeNode<K,V> lo = null, loTail = null;
  117.                         TreeNode<K,V> hi = null, hiTail = null;
  118.                         int lc = 0, hc = 0;
  119.                         for (Node<K,V> e = t.first; e != null; e = e.next) {
  120.                             int h = e.hash;
  121.                             TreeNode<K,V> p = new TreeNode<K,V>
  122.                                 (h, e.key, e.val, null, null);
  123.                             if ((h & n) == 0) {
  124.                                 if ((p.prev = loTail) == null)
  125.                                     lo = p;
  126.                                 else
  127.                                     loTail.next = p;
  128.                                 loTail = p;
  129.                                 ++lc;
  130.                             }
  131.                             else {
  132.                                 if ((p.prev = hiTail) == null)
  133.                                     hi = p;
  134.                                 else
  135.                                     hiTail.next = p;
  136.                                 hiTail = p;
  137.                                 ++hc;
  138.                             }
  139.                         }
  140.                         ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) :
  141.                             (hc != 0) ? new TreeBin<K,V>(lo) : t;
  142.                         hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) :
  143.                             (lc != 0) ? new TreeBin<K,V>(hi) : t;
  144.                         setTabAt(nextTab, i, ln);
  145.                         setTabAt(nextTab, i + n, hn);
  146.                         setTabAt(tab, i, fwd);
  147.                         advance = true;
  148.                     }
  149.                 }
  150.             }
  151.         }
  152.     }
  153. }

2、图解

微信截图_20200516213316.png

九、jdk1.8多线程扩容效率改进

多线程协助扩容的操作会在两个地方被触发：
① 当添加元素时，发现添加的元素对用的桶位为fwd节点，就会先去协助扩容，然后再添加元素
② 当添加完元素后，判断当前元素个数达到了扩容阈值，此时发现sizeCtl的值小于0，并且新数组不为空，这个时候，会去协助扩容

1、源码分析

1.1 元素未添加，先协助扩容，扩容完后再添加元素

  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 (Node<K,V>[] tab = table;;) {
  6.         Node<K,V> f; int n, i, fh;
  7.         if (tab == null || (n = tab.length) == 0)
  8.             tab = initTable();
  9.         else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
  10.             if (casTabAt(tab, i, null,
  11.                          new Node<K,V>(hash, key, value, null)))
  12.                 break;                   // no lock when adding to empty bin
  13.         }
  14.         //发现此处为fwd节点，协助扩容，扩容结束后，再循环回来添加元素
  15.         else if ((fh = f.hash) == MOVED)
  16.             tab = helpTransfer(tab, f);
  18.         //省略代码
  21. final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
  22.     Node<K,V>[] nextTab; int sc;
  23.     if (tab != null && (f instanceof ForwardingNode) &&
  24.         (nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) {
  25.         int rs = resizeStamp(tab.length);
  26.         while (nextTab == nextTable && table == tab &&
  27.                (sc = sizeCtl) < 0) {
  28.             if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
  29.                 sc == rs + MAX_RESIZERS || transferIndex <= 0)
  30.                 break;
  31.             if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
  32.                 //扩容，传递一个不是null的nextTab
  33.                 transfer(tab, nextTab);
  34.                 break;
  35.             }
  36.         }
  37.         return nextTab;
  38.     }
  39.     return table;
  40. }

1.2 先添加元素，再协助扩容

  1. private final void addCount(long x, int check) {
  2.     //省略代码
  4.     if (check >= 0) {
  5.         Node<K,V>[] tab, nt; int n, sc;
  6.           //元素个数达到扩容阈值
  7.         while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
  8.                (n = tab.length) < MAXIMUM_CAPACITY) {
  9.             int rs = resizeStamp(n);
  10.             //sizeCtl小于0，说明正在执行扩容，那么协助扩容
  11.             if (sc < 0) {
  12.                 if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
  13.                     sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
  14.                     transferIndex <= 0)
  15.                     break;
  16.                 if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
  17.                     transfer(tab, nt);
  18.             }
  19.             else if (U.compareAndSwapInt(this, SIZECTL, sc,
  20.                                          (rs << RESIZE_STAMP_SHIFT) + 2))
  21.                 transfer(tab, null);
  22.             s = sumCount();
  23.         }
  24.     }
  25. }

注意：扩容的代码都在transfer方法中，这里不再赘述

2、图解

微信截图_20200516215323.png

十、集合长度的累计方式

1、源码分析

1.1 addCount方法

① CounterCell数组不为空，优先利用数组中的CounterCell记录数量
② 如果数组为空，尝试对baseCount进行累加，失败后，会执行fullAddCount逻辑
③ 如果是添加元素操作，会继续判断是否需要扩容

  1. private final void addCount(long x, int check) {
  2.     CounterCell[] as; long b, s;
  3.     //当CounterCell数组不为空，则优先利用数组中的CounterCell记录数量
  4.     //或者当baseCount的累加操作失败，会利用数组中的CounterCell记录数量
  5.     if ((as = counterCells) != null ||
  6.         !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
  7.         CounterCell a; long v; int m;
  8.         //标识是否有多线程竞争
  9.         boolean uncontended = true;
  10.         //当as数组为空
  11.         //或者当as长度为0
  12.         //或者当前线程对应的as数组桶位的元素为空
  13.         //或者当前线程对应的as数组桶位不为空，但是累加失败
  14.         if (as == null || (m = as.length - 1) < 0 ||
  15.             (a = as[ThreadLocalRandom.getProbe() & m]) == null ||
  16.             !(uncontended =
  17.               U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
  18.             //以上任何一种情况成立，都会进入该方法，传入的uncontended是false
  19.             fullAddCount(x, uncontended);
  20.             return;
  21.         }
  22.         if (check <= 1)
  23.             return;
  24.         //计算元素个数
  25.         s = sumCount();
  26.     }
  27.     if (check >= 0) {
  28.         Node<K,V>[] tab, nt; int n, sc;
  29.         //当元素个数达到扩容阈值
  30.         //并且数组不为空
  31.         //并且数组长度小于限定的最大值
  32.         //满足以上所有条件，执行扩容
  33.         while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&
  34.                (n = tab.length) < MAXIMUM_CAPACITY) {
  35.             //这个是一个很大的正数
  36.             int rs = resizeStamp(n);
  37.             //sc小于0，说明有线程正在扩容，那么会协助扩容
  38.             if (sc < 0) {
  39.                 //扩容结束或者扩容线程数达到最大值或者扩容后的数组为null或者没有更多的桶位需要转移，结束操作
  40.                 if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
  41.                     sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||
  42.                     transferIndex <= 0)
  43.                     break;
  44.                 //扩容线程加1，成功后，进行协助扩容操作
  45.                 if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
  46.                     //协助扩容，newTable不为null
  47.                     transfer(tab, nt);
  48.             }
  49.             //没有其他线程在进行扩容，达到扩容阈值后，给sizeCtl赋了一个很大的负数
  50.             //1+1=2 --》 代表此时有一个线程在扩容
  52.             //rs << RESIZE_STAMP_SHIFT)是一个很大的负数
  53.             else if (U.compareAndSwapInt(this, SIZECTL, sc,
  54.                                          (rs << RESIZE_STAMP_SHIFT) + 2))
  55.                 //扩容，newTable为null
  56.                 transfer(tab, null);
  57.             s = sumCount();
  58.         }
  59.     }
  60. }

1.2 fullAddCount方法

① 当CounterCell数组不为空，优先对CounterCell数组中的CounterCell的value累加
② 当CounterCell数组为空，会去创建CounterCell数组，默认长度为2，并对数组中的CounterCell的value累加
③ 当数组为空，并且此时有别的线程正在创建数组，那么尝试对baseCount做累加，成功即返回，否则自旋

  1. private final void fullAddCount(long x, boolean wasUncontended) {
  2.     int h;
  3.     //获取当前线程的hash值
  4.     if ((h = ThreadLocalRandom.getProbe()) == 0) {
  5.         ThreadLocalRandom.localInit();      // force initialization
  6.         h = ThreadLocalRandom.getProbe();
  7.         wasUncontended = true;
  8.     }
  9.     //标识是否有冲突，如果最后一个桶不是null，那么为true
  10.     boolean collide = false;                // True if last slot nonempty
  11.     for (;;) {
  12.         CounterCell[] as; CounterCell a; int n; long v;
  13.         //数组不为空，优先对数组中CouterCell的value累加
  14.         if ((as = counterCells) != null && (n = as.length) > 0) {
  15.             //线程对应的桶位为null
  16.             if ((a = as[(n - 1) & h]) == null) {
  17.                 if (cellsBusy == 0) {            // Try to attach new Cell
  18.                     //创建CounterCell对象
  19.                     CounterCell r = new CounterCell(x); // Optimistic create
  20.                     //利用CAS修改cellBusy状态为1，成功则将刚才创建的CounterCell对象放入数组中
  21.                     if (cellsBusy == 0 &&
  22.                         U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
  23.                         boolean created = false;
  24.                         try {               // Recheck under lock
  25.                             CounterCell[] rs; int m, j;
  26.                             //桶位为空， 将CounterCell对象放入数组
  27.                             if ((rs = counterCells) != null &&
  28.                                 (m = rs.length) > 0 &&
  29.                                 rs[j = (m - 1) & h] == null) {
  30.                                 rs[j] = r;
  31.                                 //表示放入成功
  32.                                 created = true;
  33.                             }
  34.                         } finally {
  35.                             cellsBusy = 0;
  36.                         }
  37.                         if (created) //成功退出循环
  38.                             break;
  39.                         //桶位已经被别的线程放置了已给CounterCell对象，继续循环
  40.                         continue;           // Slot is now non-empty
  41.                     }
  42.                 }
  43.                 collide = false;
  44.             }
  45.             //桶位不为空，重新计算线程hash值，然后继续循环
  46.             else if (!wasUncontended)       // CAS already known to fail
  47.                 wasUncontended = true;      // Continue after rehash
  48.             //重新计算了hash值后，对应的桶位依然不为空，对value累加
  49.             //成功则结束循环
  50.             //失败则继续下面判断
  51.             else if (U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))
  52.                 break;
  53.             //数组被别的线程改变了，或者数组长度超过了可用cpu大小，重新计算线程hash值，否则继续下一个判断
  54.             else if (counterCells != as || n >= NCPU)
  55.                 collide = false;            // At max size or stale
  56.             //当没有冲突，修改为有冲突，并重新计算线程hash，继续循环
  57.             else if (!collide)
  58.                 collide = true;
  59.             //如果CounterCell的数组长度没有超过cpu核数，对数组进行两倍扩容
  60.             //并继续循环
  61.             else if (cellsBusy == 0 &&
  62.                      U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
  63.                 try {
  64.                     if (counterCells == as) {// Expand table unless stale
  65.                         CounterCell[] rs = new CounterCell[n << 1];
  66.                         for (int i = 0; i < n; ++i)
  67.                             rs[i] = as[i];
  68.                         counterCells = rs;
  69.                     }
  70.                 } finally {
  71.                     cellsBusy = 0;
  72.                 }
  73.                 collide = false;
  74.                 continue;                   // Retry with expanded table
  75.             }
  76.             h = ThreadLocalRandom.advanceProbe(h);
  77.         }
  78.         //CounterCell数组为空，并且没有线程在创建数组，修改标记，并创建数组
  79.         else if (cellsBusy == 0 && counterCells == as &&
  80.                  U.compareAndSwapInt(this, CELLSBUSY, 0, 1)) {
  81.             boolean init = false;
  82.             try {                           // Initialize table
  83.                 if (counterCells == as) {
  84.                     CounterCell[] rs = new CounterCell[2];
  85.                     rs[h & 1] = new CounterCell(x);
  86.                     counterCells = rs;
  87.                     init = true;
  88.                 }
  89.             } finally {
  90.                 cellsBusy = 0;
  91.             }
  92.             if (init)
  93.                 break;
  94.         }
  95.         //数组为空，并且有别的线程在创建数组，那么尝试对baseCount做累加，成功就退出循环，失败就继续循环
  96.         else if (U.compareAndSwapLong(this, BASECOUNT, v = baseCount, v + x))
  97.             break;                          // Fall back on using base
  98.     }
  99. }

2、图解

fullAddCount方法中，当as数组不为空的逻辑图解
微信截图_20200516185407.png

十一、jdk1.8集合长度获取

1、源码分析

1.1 size方法

  1. public int size() {
  2.     long n = sumCount();
  3.     return ((n < 0L) ? 0 :
  4.             (n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :
  5.             (int)n);
  6. }

1.2 sumCount方法

  1. final long sumCount() {
  2.     CounterCell[] as = counterCells; CounterCell a;
  3.     //获取baseCount的值
  4.     long sum = baseCount;
  5.     if (as != null) {
  6.         //遍历CounterCell数组，累加每一个CounterCell的value值
  7.         for (int i = 0; i < as.length; ++i) {
  8.             if ((a = as[i]) != null)
  9.                 sum += a.value;
  10.         }
  11.     }
  12.     return sum;
  13. }

注意：这个方法并不是线程安全的