[关闭]
@boothsun 2018-04-28T06:22:32.000000Z 字数 8553 阅读 1414

ConcurrentHashMap学习（先复习旧的知识点 待有时间再写）

Java

文章大部分内容为以下优秀博文 学习笔记：
1. 深入并发包 ConcurrentHashMap
2. Java进阶（六）从ConcurrentHashMap的演进看Java多线程核心技术
3. ConcurrentHashMap源码分析（JDK8版本）
4. 深入浅出ConcurrentHashMap1.8

前言

JDK1.7下 ConcurrentHashMap 是通过分段锁（锁是 ReentrantLock）的形式来保证线程安全的，底层采用的是数组+链表的存储结构。

JDK1.8下ConcurrentHashMap是通过CAS+Synchronized的形式来保证线程安全的，底层采用的是数组+链表+红黑树的存储结构。

JDK1.7下的ConcurrentHashMap

JDK1.7 中的ConcurrentHashMap的底层数据结构仍然是数组和链表。与HashMap不同的是，ConcurrentHashMap最外层不是一个大的数组，而是一个Segment的数组。每个Segment包含一个与HashMap数据结构差不多的链表数组。整个数据结构如下图所示：

image.png-112kB

另一张，我觉得还可以的图：

image.png-26kB

其包含两个核心静态内部类 Segment和HashEntry。

JDK1.8 下的ConcurrentHashMap实现原理

JDK1.8的实现已经抛弃了Segment分段锁机制，利用CAS+Synchronized来保证并发更新的安全性，底层采用数组+链表+红黑树的存储结构。默认情况下，hash冲突时产生的是链表，但当链表长度超过一定阈值（默认是8）时会自动将链表（寻址时间复杂度为O(n) ）转换为红黑树（寻址时间复杂度为O() ）。其数据结构如下图，所示：

image.png-72.7kB

另一张，我觉得还可以的图：

image.png-13.1kB

所以，整个ConcurrentHashMap看起来就像是优化过且线程安全的HashMap，虽然在JDK1.8中还能看到Segment的数据结构，但是已经是简化了属性，只是为了兼容就版本。

CurrentHashMap中的基本属性

  1. // node数组最大容量：2^30=1073741824
  2. private static final int MAXIMUM_CAPACITY = 1 << 30;
  3. // 默认初始值，必须是2的幕数
  4. private static final int DEFAULT_CAPACITY = 16;
  5. //数组可能最大值，需要与toArray（）相关方法关联
  6. static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;
  7. //并发级别，遗留下来的，为兼容以前的版本
  8. private static final int DEFAULT_CONCURRENCY_LEVEL = 16;
  9. // 负载因子
  10. private static final float LOAD_FACTOR = 0.75f;
  11. // 链表转红黑树阀值,> 8 链表转换为红黑树
  12. static final int TREEIFY_THRESHOLD = 8;
  13. //树转链表阀值，小于等于6（tranfer时，lc、hc=0两个计数器分别++记录原bin、新binTreeNode数量，<=UNTREEIFY_THRESHOLD 则untreeify(lo)）
  14. static final int UNTREEIFY_THRESHOLD = 6;
  15. static final int MIN_TREEIFY_CAPACITY = 64;
  16. private static final int MIN_TRANSFER_STRIDE = 16;
  17. private static int RESIZE_STAMP_BITS = 16;
  18. // 2^15-1，help resize的最大线程数
  19. private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1;
  20. // 32-16=16，sizeCtl中记录size大小的偏移量
  21. private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;
  22. // forwarding nodes的hash值
  23. static final int MOVED     = -1; 
  24. // 树根节点的hash值
  25. static final int TREEBIN   = -2; 
  26. // ReservationNode的hash值
  27. static final int RESERVED  = -3; 
  28. // 可用处理器数量
  29. static final int NCPU = Runtime.getRuntime().availableProcessors();
  30. //存放node的数组
  31. transient volatile Node<K,V>[] table;
  32. /* 
  33.  * 控制标识符，用来控制table的初始化和扩容的操作，不同的值有不同的含义
  34.  * 当为负数时：-1代表正在初始化，-N代表有N-1个线程正在 进行扩容
  35.  * 当为0时：代表当时的table还没有被初始化
  36.  * 当为正数时：表示初始化或者下一次进行扩容的大小
  37. private transient volatile int sizeCtl;

基本属性定义了ConcurrentHashMap的一些边界以及操作时的一些控制标志。下面，我们看一下内部的一些结构组成，这些是ConcurrentHashMap整个数据结构的核心。

Node

Node是ConcurrentHashMap存储结构的基本单元，继承于HashMap中的Entry，用于存储数据，源代码如下：

  1. static class Node<K,V> implements Map.Entry<K,V> {
  2.     //链表的数据结构
  3.     final int hash;
  4.     final K key;
  5.     //val和next都会在扩容时发生变化，所以加上volatile来保持可见性和禁止重排序
  6.     volatile V val;
  7.     volatile Node<K,V> next;
  8.     Node(int hash, K key, V val, Node<K,V> next) {
  9.         this.hash = hash;
  10.         this.key = key;
  11.         this.val = val;
  12.         this.next = next;
  13.     }
  14.     public final K getKey()       { return key; }
  15.     public final V getValue()     { return val; }
  16.     public final int hashCode()   { return key.hashCode() ^ val.hashCode(); }
  17.     public final String toString(){ return key + "=" + val; }
  18.     //不允许更新value  
  19.     public final V setValue(V value) {
  20.         throw new UnsupportedOperationException();
  21.     }
  22.     public final boolean equals(Object o) {
  23.         Object k, v, u; Map.Entry<?,?> e;
  24.         return ((o instanceof Map.Entry) &&
  25.                 (k = (e = (Map.Entry<?,?>)o).getKey()) != null &&
  26.                 (v = e.getValue()) != null &&
  27.                 (k == key || k.equals(key)) &&
  28.                 (v == (u = val) || v.equals(u)));
  29.     }
  30.     //用于map中的get（）方法，子类重写
  31.     Node<K,V> find(int h, Object k) {
  32.         Node<K,V> e = this;
  33.         if (k != null) {
  34.             do {
  35.                 K ek;
  36.                 if (e.hash == h &&
  37.                     ((ek = e.key) == k || (ek != null && k.equals(ek))))
  38.                     return e;
  39.             } while ((e = e.next) != null);
  40.         }
  41.         return null;
  42.     }
  43. }

Node的数据结构很简单，它是链表中的一个节点，但是只允许对数据进行查找，不允许进行修改。

TreeNode

TreeNode继承于Node，但是数据结构换成了二叉树结构，它是红黑树的数据的存储结构，用于红黑树中存储数据，当链表的节点数大于8时会转换为红黑树的结构，此时就是通过ThreeNode作为存储结构代替Node来转换成红黑树。具体源代码如下：

  1. static final class TreeNode<K,V> extends Node<K,V> {
  2.     //树形结构的属性定义
  3.     TreeNode<K,V> parent;  // red-black tree links
  4.     TreeNode<K,V> left;
  5.     TreeNode<K,V> right;
  6.     TreeNode<K,V> prev;    // needed to unlink next upon deletion
  7.     boolean red; //标志红黑树的红节点
  8.     TreeNode(int hash, K key, V val, Node<K,V> next,
  9.              TreeNode<K,V> parent) {
  10.         super(hash, key, val, next);
  11.         this.parent = parent;
  12.     }
  13.     Node<K,V> find(int h, Object k) {
  14.         return findTreeNode(h, k, null);
  15.     }
  16.     //根据key查找 从根节点开始找出相应的TreeNode，
  17.     final TreeNode<K,V> findTreeNode(int h, Object k, Class<?> kc) {
  18.         if (k != null) {
  19.             TreeNode<K,V> p = this;
  20.             do  {
  21.                 int ph, dir; K pk; TreeNode<K,V> q;
  22.                 TreeNode<K,V> pl = p.left, pr = p.right;
  23.                 if ((ph = p.hash) > h)
  24.                     p = pl;
  25.                 else if (ph < h)
  26.                     p = pr;
  27.                 else if ((pk = p.key) == k || (pk != null && k.equals(pk)))
  28.                     return p;
  29.                 else if (pl == null)
  30.                     p = pr;
  31.                 else if (pr == null)
  32.                     p = pl;
  33.                 else if ((kc != null ||
  34.                           (kc = comparableClassFor(k)) != null) &&
  35.                          (dir = compareComparables(kc, k, pk)) != 0)
  36.                     p = (dir < 0) ? pl : pr;
  37.                 else if ((q = pr.findTreeNode(h, k, kc)) != null)
  38.                     return q;
  39.                 else
  40.                     p = pl;
  41.             } while (p != null);
  42.         }
  43.         return null;
  44.     }
  45. }

TreeBin

TreeBin可以看成是封装了操作TreeNode的工具类，它提供了一些转换红黑树的条件判断和锁控制的工具方法。

  1. static final class TreeBin<K,V> extends Node<K,V> {
  2.     //指向TreeNode列表和根节点
  3.     TreeNode<K,V> root;
  4.     volatile TreeNode<K,V> first;
  5.     volatile Thread waiter;
  6.     volatile int lockState;
  7.     // 读写锁状态
  8.     static final int WRITER = 1; // 获取写锁的状态
  9.     static final int WAITER = 2; // 等待写锁的状态
  10.     static final int READER = 4; // 增加数据时读锁的状态
  11.     /**
  12.      * 初始化红黑树
  13.      */
  14.     TreeBin(TreeNode<K,V> b) {
  15.         super(TREEBIN, null, null, null);
  16.         this.first = b;
  17.         TreeNode<K,V> r = null;
  18.         for (TreeNode<K,V> x = b, next; x != null; x = next) {
  19.             next = (TreeNode<K,V>)x.next;
  20.             x.left = x.right = null;
  21.             if (r == null) {
  22.                 x.parent = null;
  23.                 x.red = false;
  24.                 r = x;
  25.             }
  26.             else {
  27.                 K k = x.key;
  28.                 int h = x.hash;
  29.                 Class<?> kc = null;
  30.                 for (TreeNode<K,V> p = r;;) {
  31.                     int dir, ph;
  32.                     K pk = p.key;
  33.                     if ((ph = p.hash) > h)
  34.                         dir = -1;
  35.                     else if (ph < h)
  36.                         dir = 1;
  37.                     else if ((kc == null &&
  38.                               (kc = comparableClassFor(k)) == null) ||
  39.                              (dir = compareComparables(kc, k, pk)) == 0)
  40.                         TreeNode<K,V> xp = p;
  41.                     if ((p = (dir <= 0) ? p.left : p.right) == null) {
  42.                         x.parent = xp;
  43.                         if (dir <= 0)
  44.                             xp.left = x;
  45.                         else
  46.                             xp.right = x;
  47.                         r = balanceInsertion(r, x);
  48.                         break;
  49.                     }
  50.                 }
  51.             }
  52.         }
  53.         this.root = r;
  54.         assert checkInvariants(root);
  55.     }
  56.     ......
  57. }

put 操作的过程

  1. public V put(K key, V value) {
  2.     return putVal(key, value, false);
  3. }
  4. /** Implementation for put and putIfAbsent */
  5. final V putVal(K key, V value, boolean onlyIfAbsent) {
  6.     if (key == null || value == null) throw new NullPointerException();
  7.     int hash = spread(key.hashCode()); //两次hash，减少hash冲突，可以均匀分布
  8.     int binCount = 0;
  9.     for (Node<K,V>[] tab = table;;) { //对这个table进行迭代
  10.         Node<K,V> f; int n, i, fh;
  11.         //这里就是上面构造方法没有进行初始化，在这里进行判断，为null就调用initTable进行初始化，属于懒汉模式初始化
  12.         if (tab == null || (n = tab.length) == 0)
  13.             tab = initTable();
  14.         else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {//如果i位置没有数据，就直接无锁插入
  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.         else if ((fh = f.hash) == MOVED)//如果在进行扩容，则先进行扩容操作
  20.             tab = helpTransfer(tab, f);
  21.         else {
  22.             V oldVal = null;
  23.             //如果以上条件都不满足，那就要进行加锁操作，也就是存在hash冲突，锁住链表或者红黑树的头结点
  24.             synchronized (f) {
  25.                 if (tabAt(tab, i) == f) {
  26.                     if (fh >= 0) { //表示该节点是链表结构
  27.                         binCount = 1;
  28.                         for (Node<K,V> e = f;; ++binCount) {
  29.                             K ek;
  30.                             //这里涉及到相同的key进行put就会覆盖原先的value
  31.                             if (e.hash == hash &&
  32.                                 ((ek = e.key) == key ||
  33.                                  (ek != null && key.equals(ek)))) {
  34.                                 oldVal = e.val;
  35.                                 if (!onlyIfAbsent)
  36.                                     e.val = value;
  37.                                 break;
  38.                             }
  39.                             Node<K,V> pred = e;
  40.                             if ((e = e.next) == null) {  //插入链表尾部
  41.                                 pred.next = new Node<K,V>(hash, key,
  42.                                                           value, null);
  43.                                 break;
  44.                             }
  45.                         }
  46.                     }
  47.                     else if (f instanceof TreeBin) {//红黑树结构
  48.                         Node<K,V> p;
  49.                         binCount = 2;
  50.                         //红黑树结构旋转插入
  51.                         if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
  52.                                                        value)) != null) {
  53.                             oldVal = p.val;
  54.                             if (!onlyIfAbsent)
  55.                                 p.val = value;
  56.                         }
  57.                     }
  58.                 }
  59.             }
  60.             if (binCount != 0) { //如果链表的长度大于8时就会进行红黑树的转换
  61.                 if (binCount >= TREEIFY_THRESHOLD)
  62.                     treeifyBin(tab, i);
  63.                 if (oldVal != null)
  64.                     return oldVal;
  65.                 break;
  66.             }
  67.         }
  68.     }
  69.     addCount(1L, binCount);//统计size，并且检查是否需要扩容
  70.     return null;
  71. }

这个put的过程很清晰，首先会进入无条件自循直到put成功，可以用以下六步流程来概括：

  1. 如果没有初始化就先调用initTable()方法来进行初始化过程。
  2. 如果没有hash冲突就直接CAS插入。
  3. 如果还在进行扩容操作就先进行扩容。
  4. 如果存在Hash冲突，就加锁来保证线程安全，这里有两种情况，一种是链表形式就直接遍历到尾端插入，一种是红黑树就按照红黑树结构插入。
  5. 最后一个如果该链表的数量大于阈值8，就要先转换成红黑树的结构。
  6. 如果添加成功就调用addCount()方法统计size，并且检查是否需要扩容。

现在我们来对每一步的细节进行源码分析，在第一步中，符合条件会进行初始化操作，我们来看看initTable()方法：

  1. /**
  2.  * Initializes table, using the size recorded in sizeCtl.
  3.  */
  4. private final Node<K,V>[] initTable() {
  5.     Node<K,V>[] tab; int sc;
  6.     while ((tab = table) == null || tab.length == 0) {//空的table才能进入初始化操作
  7.         if ((sc = sizeCtl) < 0) //sizeCtl<0表示其他线程已经在初始化了或者扩容了，挂起当前线程 
  8.             Thread.yield(); // lost initialization race; just spin
  9.         else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {//CAS操作SIZECTL为-1，表示初始化状态
  10.             try {
  11.                 if ((tab = table) == null || tab.length == 0) {
  12.                     int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
  13.                     @SuppressWarnings("unchecked")
  14.                     Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];//初始化
  15.                     table = tab = nt;
  16.                     sc = n - (n >>> 2);//记录下次扩容的大小
  17.                 }
  18.             } finally {
  19.                 sizeCtl = sc;
  20.             }
  21.             break;
  22.         }
  23.     }
  24.     return tab;
  25. }

上面的初始化过程非常简单，首先会验证sizeCtl的值是否小于0，如果是，则表示其他线程正在进行扩容或者已经在初始化了，则会挂起当前线程。否则，则表明当前还没有进行过初始化table的过程，就会进入初始化流程。初始化流程也非常简单，就是new了指定大小的Node数组，然后计算下次扩容的大小。

在第二步中没有hash冲突就直接调用Unsafe的方法CAS插入该元素，这步很简单，没有太多可说的。

第三步，如果容器正在扩容，则会调用helpTransfer()方法帮助扩容，现在我们来看看helpTransfer()方法看看。

  1. /**
  2.  * 帮助从旧的table的元素复制到新的table中
  3.  */
  4. final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
  5.     Node<K,V>[] nextTab; int sc;
  6.     if (tab != null && (f instanceof ForwardingNode) &&
  7.         (nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) { //新的table nextTba已经存在前提下才能帮助扩容
  8.         int rs = resizeStamp(tab.length);
  9.         while (nextTab == nextTable && table == tab &&
  10.                (sc = sizeCtl) < 0) {
  11.             if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
  12.                 sc == rs + MAX_RESIZERS || transferIndex <= 0)
  13.                 break;
  14.             if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
  15.                 transfer(tab, nextTab);//调用扩容方法
  16.                 break;
  17.             }
  18.         }
  19.         return nextTab;
  20.     }
  21.     return table;
  22. }

其实helpTransfer()方法的目的就是调用多个工作线程一起帮助进行扩容，这样的效率就会更高，而不是只有检查到要扩容的那个线程进行扩容操作，其他线程就要等待扩容操作完成才能工作。

后面继续参考写：

深入并发包 ConcurrentHashMap

ConcurrentHashMap源码分析（JDK8版本）

加锁代码段：

  1. f = tabAt(tab, i = (n - 1) & hash)
  2.  synchronized (f) {
添加新批注
在作者公开此批注前，只有你和作者可见。
回复批注