JavaConcurrentHashMap高并发安全实现原理解析

 一、概述

ConcurrentHashMap (以下简称C13Map) 是并发编程出场率最高的数据结构之一，大量的并发CASE背后都有C13Map的支持，同时也是JUC包中代码量最大的组件(6000多行)，自JDK8开始Oracle对其进行了大量优化工作。

本文从 HashMap 的基础知识开始，尝试逐一分析C13Map中各个组件的实现和安全性保证。

二、HashMap基础知识 

分析C13MAP前，需要了解以下的HashMap知识或者约定：

  •  哈希表的长度永远都是2的幂次方，原因是hashcode%tableSize==hashcode&(tableSize-1)，也就是哈希槽位的确定可以用一次与运算来替代取余运算。
  •  会对hashcode调用若干次扰动函数，将高16位与低16位做异或运算，因为高16位的随机性更强。
  •  当表中的元素总数超过tableSize * 0.75时，哈希表会发生扩容操作，每次扩容的tableSize是原先的两倍。
  •  下文提到的槽位(bucket)、哈希分桶、BIN均表示同一个概念，即哈希table上的某一列。
  •  旧表在做搬运时i槽位的node可以根据其哈希值的第tableSize位的bit决定在新表上的槽位是i还是i+tableSize。
  •  每个槽位上有可能会出现哈希冲突，在未达到某个阈值时它是一个链表结构，达到阈值后会升级到红黑树结构。
  •  HashMap本身并非为多线程环境设计，永远不要尝试在并发环境下直接使用HashMap，C13Map不存在这个安全问题。

三、C13Map的字段定义

C13Map的字段定义 

 
 
 
 
    
  
  
  
  
  1. //最大容量  
    2. 
  
  
  
  
  2. private static final int MAXIMUM_CAPACITY = 1 << 30;   
    3. 
  
  
  
  
  3. //默认初始容量  
    4. 
  
  
  
  
  4. private static final int DEFAULT_CAPACITY = 16;  
    5. 
  
  
  
  
  5. //数组的最大容量,防止抛出OOM  
    6. 
  
  
  
  
  6. static final int MAX_ARRAY_SIZE = Integer.MAX_VALUE - 8;   
    7. 
  
  
  
  
  7. //最大并行度，仅用于兼容JDK1.7以前版本  
    8. 
  
  
  
  
  8. private static final int DEFAULT_CONCURRENCY_LEVEL = 16;  
    9. 
  
  
  
  
  9. //扩容因子  
    10. 
  
  
  
  
  10. private static final float LOAD_FACTOR = 0.75f;  
    11. 
  
  
  
  
  11. //链表转红黑树的阈值  
    12. 
  
  
  
  
  12. static final int TREEIFY_THRESHOLD = 8;  
    13. 
  
  
  
  
  13. //红黑树退化阈值  
    14. 
  
  
  
  
  14. static final int UNTREEIFY_THRESHOLD = 6;  
    15. 
  
  
  
  
  15. //链表转红黑树的最小总量  
    16. 
  
  
  
  
  16. static final int MIN_TREEIFY_CAPACITY = 64;  
    17. 
  
  
  
  
  17. //扩容搬运时批量搬运的最小槽位数  
    18. 
  
  
  
  
  18. private static final int MIN_TRANSFER_STRIDE = 16;  
    19. 
  
  
  
  
  19. //当前待扩容table的邮戳位,通常是高16位  
    20. 
  
  
  
  
  20. private static final int RESIZE_STAMP_BITS = 16;  
    21. 
  
  
  
  
  21. //同时搬运的线程数自增的最大值  
    22. 
  
  
  
  
  22. private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1;  
    23. 
  
  
  
  
  23. //搬运线程数的标识位，通常是低16位  
    24. 
  
  
  
  
  24. private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;  
    25. 
  
  
  
  
  25. static final int MOVED     = -1; // 说明是forwardingNode  
    26. 
  
  
  
  
  26. static final int TREEBIN   = -2; // 红黑树  
    27. 
  
  
  
  
  27. static final int RESERVED  = -3; // 原子计算的占位Node  
    28. 
  
  
  
  
  28. static final int HASH_BITS = 0x7fffffff; // 保证hashcode扰动计算结果为正数  
    29. 
  
  
  
  
  29. //当前哈希表  
    30. 
  
  
  
  
  30. transient volatile Node[] table;  
    31. 
  
  
  
  
  31. //下一个哈希表  
    32. 
  
  
  
  
  32. private transient volatile Node[] nextTable;  
    33. 
  
  
  
  
  33. //计数的基准值  
    34. 
  
  
  
  
  34. private transient volatile long baseCount;  
    35. 
  
  
  
  
  35. //控制变量，不同场景有不同用途，参考下文  
    36. 
  
  
  
  
  36. private transient volatile int sizeCtl;  
    37. 
  
  
  
  
  37. //并发搬运过程中CAS获取区段的下限值  
    38. 
  
  
  
  
  38. private transient volatile int transferIndex;  
    39. 
  
  
  
  
  39. //计数cell初始化或者扩容时基于此字段使用自旋锁  
    40. 
  
  
  
  
  40. private transient volatile int cellsBusy;  
    41. 
  
  
  
  
  41. //加速多核CPU计数的cell数组  
    42. 
  
  
  
  
  42. private transient volatile CounterCell[] counterCells; 
    43.

四、安全操作Node数组 

 
 
 
 
    
  
  
  
  
  1. static final  Node tabAt(Node[] tab, int i) {  
    2. 
  
  
  
  
  2.     return (Node)U.getReferenceAcquire(tab, ((long)i << ASHIFT) + ABASE);  
    3. 
  
  
  
  
  3. }  
    4. 
  
  
  
  
  4. static final  boolean casTabAt(Node[] tab, int i,  
    5. 
  
  
  
  
  5.                                     Node c, Node v) {  
    6. 
  
  
  
  
  6.     return U.compareAndSetReference(tab, ((long)i << ASHIFT) + ABASE, c, v);  
    7. 
  
  
  
  
  7. }  
    8. 
  
  
  
  
  8. static final  void setTabAt(Node[] tab, int i, Node v) {  
    9. 
  
  
  
  
  9.     U.putReferenceRelease(tab, ((long)i << ASHIFT) + ABASE, v); 
    10. 
  
  
  
  
  10. } 
    11.

对Node[] 上任意一个index的读取和写入都使用了Unsafe辅助类，table本身是volatile类型的并不能保证其下的每个元素的内存语义也是volatile类型；

需要借助于Unsafe来保证Node[]元素的“读/写/CAS”操作在多核并发环境下的原子或者可见性。

五、读操作get为什么是线程安全的

首先需要明确的是，C13Map的读操作一般是不加锁的(TreeBin的读写锁除外)，而读操作与写操作有可能并行；可以保证的是，因为C13Map的写操作都要获取bin头部的syncronized互斥锁，能保证最多只有一个线程在做更新，这其实是一个单线程写、多线程读的并发安全性的问题。

C13Map的get方法 

 
 
 
 
    
  
  
  
  
  1. public V get(Object key) {  
    2. 
  
  
  
  
  2.     Node[] tab; Node e, p; int n, eh; K ek;  
    3. 
  
  
  
  
  3.     //执行扰动函数  
    4. 
  
  
  
  
  4.     int h = spread(key.hashCode());  
    5. 
  
  
  
  
  5.     if ((tab = table) != null && (n = tab.length) > 0 && (e = tabAt(tab, (n - 1) & h)) != null) {  
    6. 
  
  
  
  
  6.         if ((eeh = e.hash) == h) {  
    7. 
  
  
  
  
  7.             if ((eek = e.key) == key || (ek != null && key.equals(ek)))  
    8. 
  
  
  
  
  8.                 return e.val;  
    9. 
  
  
  
  
  9.         } 
    10. 
  
  
  
  
  10.          else if (eh < 0)  
    11. 
  
  
  
  
  11.             return (p = e.find(h, key)) != null ? p.val : null;  
    12. 
  
  
  
  
  12.         while ((ee = e.next) != null) {  
    13. 
  
  
  
  
  13.             if (e.hash == h &&  
    14. 
  
  
  
  
  14.                 ((eek = e.key) == key || (ek != null && key.equals(ek))))  
    15. 
  
  
  
  
  15.                 return e.val; 
    16. 
  
  
  
  
  16.          }  
    17. 
  
  
  
  
  17.     }  
    18. 
  
  
  
  
  18.     return null;  
    19. 
  
  
  
  
  19. } 
    20.

1、如果当前哈希表table为null

哈希表未初始化或者正在初始化未完成，直接返回null；虽然line5和line18之间其它线程可能经历了千山万水，至少在判断tab==null的时间点key肯定是不存在的，返回null符合某一时刻的客观事实。

2、如果读取的bin头节点为null

说明该槽位尚未有节点，直接返回null。

3、如果读取的bin是一个链表

说明头节点是个普通Node。

（1）如果正在发生链表向红黑树的treeify工作，因为treeify本身并不破坏旧的链表bin的结构，只是在全部treeify完成后将头节点一次性替换为新创建的TreeBin，可以放心读取。

（2）如果正在发生resize且当前bin正在被transfer，因为transfer本身并不破坏旧的链表bin的结构，只是在全部transfer完成后将头节点一次性替换为ForwardingNode，可以放心读取。

（3）如果其它线程正在操作链表，在当前线程遍历链表的任意一个时间点，都有可能同时在发生add/replace/remove操作。

  •  如果是add操作，因为链表的节点新增从JDK8以后都采用了后入式，无非是多遍历或者少遍历一个tailNode。
  •  如果是remove操作，存在遍历到某个Node时，正好有其它线程将其remove，导致其孤立于整个链表之外；但因为其next引用未发生变更，整个链表并没有断开，还是可以照常遍历链表直到tailNode。
  •  如果是replace操作，链表的结构未变，只是某个Node的value发生了变化，没有安全问题。

结论：对于链表这种线性数据结构，单线程写且插入操作保证是后入式的前提下，并发读取是安全的；不会存在误读、链表断开导致的漏读、读到环状链表等问题。

4、如果读取的bin是一个红黑树

说明头节点是个TreeBin节点。

（1）如果正在发生红黑树向链表的untreeify操作，因为untreeify本身并不破坏旧的红黑树结构，只是在全部untreeify完成后将头节点一次性替换为新创建的普通Node，可以放心读取。

（2）如果正在发生resize且当前bin正在被transfer，因为transfer本身并不破坏旧的红黑树结构，只是在全部transfer完成后将头节点一次性替换为ForwardingNode，可以放心读取。

（3）如果其他线程在操作红黑树，在当前线程遍历红黑树的任意一个时间点，都可能有单个的其它线程发生add/replace/remove/红黑树的翻转等操作，参考下面的红黑树的读写锁实现。

TreeBin中的读写锁实现 

 
 
 
 
    
  
  
  
  
  1.  TreeNode root;  
    2. 
  
  
  
  
  2.     volatile TreeNode first;  
    3. 
  
  
  
  
  3.     volatile Thread waiter;  
    4. 
  
  
  
  
  4.     volatile int lockState;  
    5. 
  
  
  
  
  5.     // values for lockState  
    6. 
  
  
  
  
  6.     static final int WRITER = 1; // set while holding write lock  
    7. 
  
  
  
  
  7.     static final int WAITER = 2; // set when waiting for write lock  
    8. 
  
  
  
  
  8.     static final int READER = 4; // increment value for setting read lock  
    9. 
  
  
  
  
  9.     private final void lockRoot() {  
    10. 
  
  
  
  
  10.         //如果一次性获取写锁失败，进入contendedLock循环体，循环获取写锁或者休眠等待  
    11. 
  
  
  
  
  11.         if (!U.compareAndSetInt(this, LOCKSTATE, 0, WRITER))  
    12. 
  
  
  
  
  12.             contendedLock(); // offload to separate method  
    13. 
  
  
  
  
  13.     }  
    14. 
  
  
  
  
  14.     private final void unlockRoot() {  
    15. 
  
  
  
  
  15.         lockState = 0; 
    16. 
  
  
  
  
  16.     }  
    17. 
  
  
  
  
  17.     //对红黑树加互斥锁,也就是写锁  
    18. 
  
  
  
  
  18.     private final void contendedLock() {  
    19. 
  
  
  
  
  19.         boolean waiting = false;  
    20. 
  
  
  
  
  20.         for (int s;;) {  
    21. 
  
  
  
  
  21.             //如果lockState除了第二位外其它位上都为0，表示红黑树当前既没有上读锁，又没有上写锁，仅有可能存在waiter，可以尝试直接获取写锁  
    22. 
  
  
  
  
  22.             if (((s = lockState) & ~WAITER) == 0) {  
    23. 
  
  
  
  
  23.                 if (U.compareAndSetInt(this, LOCKSTATE, s, WRITER)) { 
    24. 
  
  
  
  
  24.                      if (waiting)  
    25. 
  
  
  
  
  25.                         waiter = null;  
    26. 
  
  
  
  
  26.                     return; 
    27. 
  
  
  
  
  27.                 }  
    28. 
  
  
  
  
  28.             }  
    29. 
  
  
  
  
  29.             //如果lockState第二位是0,表示当前没有线程在等待写锁  
    30. 
  
  
  
  
  30.             else if ((s & WAITER) == 0) {  
    31. 
  
  
  
  
  31.                 //将lockState的第二位设置为1，相当于打上了waiter的标记，表示有线程在等待写锁  
    32. 
  
  
  
  
  32.                 if (U.compareAndSetInt(this, LOCKSTATE, s, s | WAITER)) {  
    33. 
  
  
  
  
  33.                     waiting = true;  
    34. 
  
  
  
  
  34.                     waiter = Thread.currentThread();  
    35. 
  
  
  
  
  35.                 }  
    36. 
  
  
  
  
  36.             }  
    37. 
  
  
  
  
  37.             //休眠当前线程  
    38. 
  
  
  
  
  38.             else if (waiting)  
    39. 
  
  
  
  
  39.                 LockSupport.park(this);  
    40. 
  
  
  
  
  40.         }  
    41. 
  
  
  
  
  41.     } 
    42. 
  
  
  
  
  42.      //查找红黑树中的某个节点  
    43. 
  
  
  
  
  43.     final Node find(int h, Object k) {  
    44. 
  
  
  
  
  44.         if (k != null) {  
    45. 
  
  
  
  
  45.             for (Node e = first; e != null; ) {  
    46. 
  
  
  
  
  46.                 int s; K ek;  
    47. 
  
  
  
  
  47.                 //如果当前有waiter或者有写锁，走线性检索，因为红黑树虽然替代了链表，但其内部依然保留了链表的结构，虽然链表的查询性能一般，但根据先前的分析其读取的安全性有保证。  
    48. 
  
  
  
  
  48.                 //发现有写锁改走线性检索，是为了避免等待写锁释放花去太久时间; 而发现有waiter改走线性检索，是为了避免读锁叠加的太多，导致写锁线程需要等待太长的时间; 本质上都是为了减少读写碰撞 
    49. 
  
  
  
  
  49.                  //线性遍历的过程中，每遍历到下一个节点都做一次判断，一旦发现锁竞争的可能性减少就改走tree检索以提高性能 
    50. 
  
  
  
  
  50.                  if (((s = lockState) & (WAITER|WRITER)) != 0) {  
    51. 
  
  
  
  
  51.                     if (e.hash == h &&  
    52. 
  
  
  
  
  52.                         ((eek = e.key) == k || (ek != null && k.equals(ek))))  
    53. 
  
  
  
  
  53.                         return e;  
    54. 
  
  
  
  
  54.                     ee = e.next;  
    55. 
  
  
  
  
  55.                 }  
    56. 
  
  
  
  
  56.                 //对红黑树加共享锁,也就是读锁,CAS一次性增加4，也就是增加的只是3~32位  
    57. 
  
  
  
  
  57.                 else if (U.compareAndSetInt(this, LOCKSTATE, s, 
    58. 
  
  
  
  
  58.                                               s + READER)) {  
    59. 
  
  
  
  
  59.                     TreeNode r, p;  
    60. 
  
  
  
  
  60.                     try {  
    61. 
  
  
  
  
  61.                         p = ((r = root) == null ? null :  
    62. 
  
  
  
  
  62.                              r.findTreeNode(h, k, null));  
    63. 
  
  
  
  
  63.                     } finally {  
    64. 
  
  
  
  
  64.                         Thread w;  
    65. 
  
  
  
  
  65.                         //释放读锁，如果释放完毕且有waiter,则将其唤醒  
    66. 
  
  
  
  
  66.                         if (U.getAndAddInt(this, LOCKSTATE, -READER) ==  
    67. 
  
  
  
  
  67.                             (READER|WAITER) && (w = waiter) != null)  
    68. 
  
  
  
  
  68.                             LockSupport.unpark(w);  
    69. 
  
  
  
  
  69.                     }  
    70. 
  
  
  
  
  70.                     return p;  
    71. 
  
  
  
  
  71.                 }  
    72. 
  
  
  
  
  72.             }  
    73. 
  
  
  
  
  73.         }  
    74. 
  
  
  
  
  74.         return null;  
    75. 
  
  
  
  
  75.     }  
    76. 
  
  
  
  
  76.     //更新红黑树中的某个节点  
    77. 
  
  
  
  
  77.     final TreeNode putTreeVal(int h, K k, V v) {  
    78. 
  
  
  
  
  78.         Class kc = null;  
    79. 
  
  
  
  
  79.         boolean searched = false;  
    80. 
  
  
  
  
  80.         for (TreeNode p = root;;) {  
    81. 
  
  
  
  
  81.             int dir, ph; K pk;  
    82. 
  
  
  
  
  82.             //...省略处理红黑树数据结构的代码若干        
    83. 
  
  
  
  
  83.                  else {  
    84. 
  
  
  
  
  84.                     //写操作前加互斥锁  
    85. 
  
  
  
  
  85.                     lockRoot();  
    86. 
  
  
  
  
  86.                     try {  
    87. 
  
  
  
  
  87.                         root = balanceInsertion(root, x);  
    88. 
  
  
  
  
  88.                     } finally {  
    89. 
  
  
  
  
  89.                         //释放互斥锁 
    90. 
  
  
  
  
  90.                          unlockRoot();  
    91. 
  
  
  
  
  91.                     }  
    92. 
  
  
  
  
  92.                 }  
    93. 
  
  
  
  
  93.                 break;  
    94. 
  
  
  
  
  94.             }  
    95. 
  
  
  
  
  95.         }  
    96. 
  
  
  
  
  96.         assert checkInvariants(root);  
    97. 
  
  
  
  
  97.         return null;  
    98. 
  
  
  
  
  98.     }  
    99. 
  
  
  
  
  99. } 
    100.

红黑树内置了一套读写锁的逻辑，其内部定义了32位的int型变量lockState，第1位是写锁标志位，第2位是写锁等待标志位，从3~32位则是共享锁标志位。

读写操作是互斥的，允许多个线程同时读取，但不允许读写操作并行，同一时刻只允许一个线程进行写操作；这样任意时间点读取的都是一个合法的红黑树，整体上是安全的。

有的同学会产生疑惑，写锁释放时为何没有将waiter唤醒的操作呢？是否有可能A线程进入了等待区，B线程获取了写锁，释放写锁时仅做了lockState=0的操作。

那么A线程是否就没有机会被唤醒了，只有等待下一个读锁释放时的唤醒了呢 ？

显然这种情况违背常理，C13Map不会出现这样的疏漏，再进一步观察，红黑树的变更操作的外围，也就是在putValue/replaceNode那一层，都是对BIN的头节点加了synchornized互斥锁的，同一时刻只能有一个写线程进入TreeBin的方法范围内，当写线程发现当前waiter不为空，其实此waiter只能是当前线程自己，可以放心的获取写锁，不用担心无法被唤醒的问题。

TreeBin在find读操作检索时，在linearSearch(线性检索)和treeSearch(树检索)间做了折衷，前者性能差但并发安全，后者性能佳但要做并发控制，可能导致锁竞争；设计者使用线性检索来尽量避免读写碰撞导致的锁竞争，但评估到race condition已消失时，又立即趋向于改用树检索来提高性能，在安全和性能之间做到了极佳的平衡。具体的折衷策略请参考find方法及注释。

由于有线性检索这样一个抄底方案，以及入口处bin头节点的synchornized机制，保证了进入到TreeBin整体代码块的写线程只有一个；TreeBin中读写锁的整体设计与ReentrantReadWriteLock相比还是简单了不少，比如并未定义用于存放待唤醒线程的threadQueue，以及读线程仅会自旋而不会阻塞等等， 可以看做是特定条件下ReadWriteLock的简化版本。

5、如果读取的bin是一个ForwardingNode

说明当前bin已迁移，调用其find方法到nextTable读取数据。

forwardingNode的find方法 

 
 
 
 
    
  
  
  
  
  1. static final class ForwardingNode extends Node {  
    2. 
  
  
  
  
  2.     final Node[] nextTable;  
    3. 
  
  
  
  
  3.     ForwardingNode(Node[] tab) {  
    4. 
  
  
  
  
  4.         super(MOVED, null, null);  
    5. 
  
  
  
  
  5.         this.nextTable = tab;  
    6. 
  
  
  
  
  6.     }  
    7. 
  
  
  
  
  7.      //递归检索哈希表链  
    8. 
  
  
  
  
  8.     Node find(int h, Object k) {  
    9. 
  
  
  
  
  9.         // loop to avoid arbitrarily deep recursion on forwarding nodes  
    10. 
  
  
  
  
  10.         outer: for (Node[] tab = nextTable;;) {  
    11. 
  
  
  
  
  11.             Node e; int n;  
    12. 
  
  
  
  
  12.             if (k == null || tab == null || (n = tab.length) == 0 ||  
    13. 
  
  
  
  
  13.                 (e = tabAt(tab, (n - 1) & h)) == null)  
    14. 
  
  
  
  
  14.                 return null;  
    15. 
  
  
  
  
  15.             for (;;) {  
    16. 
  
  
  
  
  16.                 int eh; K ek; 
    17. 
  
  
  
  
  17.                  if ((eeh = e.hash) == h &&  
    18. 
  
  
  
  
  18.                     ((eek = e.key) == k || (ek != null && k.equals(ek))))  
    19. 
  
  
  
  
  19.                     return e;  
    20. 
  
  
  
  
  20.                 if (eh < 0) {  
    21. 
  
  
  
  
  21.                     if (e instanceof ForwardingNode) {  
    22. 
  
  
  
  
  22.                         tab = ((ForwardingNode)e).nextTable;  
    23. 
  
  
  
  
  23.                         continue outer;  
    24. 
  
  
  
  
  24.                     }  
    25. 
  
  
  
  
  25.                     else  
    26. 
  
  
  
  
  26.                         return e.find(h, k);  
    27. 
  
  
  
  
  27.                 }  
    28. 
  
  
  
  
  28.                 if ((ee = e.next) == null)  
    29. 
  
  
  
  
  29.                     return null;  
    30. 
  
  
  
  
  30.             }  
    31. 
  
  
  
  
  31.         }  
    32. 
  
  
  
  
  32.     }  
    33. 
  
  
  
  
  33. } 
    34.

ForwardingNode中保存了nextTable的引用，会转向下一个哈希表进行检索，但并不能保证nextTable就一定是currentTable，因为在高并发插入的情况下，极短时间内就可以导致哈希表的多次扩容，内存中极有可能驻留一条哈希表链，彼此以bin的头节点上的ForwardingNode相连，线程刚读取时拿到的是table1，遍历时却有可能经历了哈希表的链条。

eh<0有三种情况：

  •  如果是ForwardingNode继续遍历下一个哈希表。
  •  如果是TreeBin,调用其find方法进入TreeBin读写锁的保护区读取数据。
  •  如果是ReserveNode，说明当前有compute计算中，整条bin还是一个空结构，直接返回null。

6、如果读取的bin是一个ReserveNode

ReserveNode用于compute/computeIfAbsent原子计算的方法，在BIN的头节点为null且计算尚未完成时，先在bin的头节点打上一个ReserveNode的占位标记。

读操作发现ReserveNode直接返回null，写操作会因为争夺ReserveNode的互斥锁而进入阻塞态，在compute完成后被唤醒后循环重试。

六、写操作putValue/replaceNode为什么是线程安全的

典型的编程范式如下：

C13Map的putValue方法 

 
 
 
 
    
  
  
  
  
  1. Node[] tab = table;  //将堆中的table变量赋给线程堆栈中的局部变量  
    2. 
  
  
  
  
  2. Node f = tabAt(tab, i );  
    3. 
  
  
  
  
  3. if(f==null){  
    4. 
  
  
  
  
  4.  //当前槽位没有头节点,直接CAS写入  
    5. 
  
  
  
  
  5.  if (casTabAt(tab, i, null, new Node(hash, key, value)))  
    6. 
  
  
  
  
  6.     break;  
    7. 
  
  
  
  
  7. }else if(f.hash == MOVED){  
    8. 
  
  
  
  
  8.  //加入协助搬运行列  
    9. 
  
  
  
  
  9.  helpTransfer(tab,f);  
    10. 
  
  
  
  
  10. }  
    11. 
  
  
  
  
  11. //不是forwardingNode  
    12. 
  
  
  
  
  12. else if(f.hash != MOVED){  
    13. 
  
  
  
  
  13.     //先锁住I槽位上的头节点  
    14. 
  
  
  
  
  14.     synchronized (f) {  
    15. 
  
  
  
  
  15.     //再doubleCheck看此槽位上的头节点是否还是f  
    16. 
  
  
  
  
  16.     if (tabAt(tab, i) == f) {  
    17. 
  
  
  
  
  17.        ...各种写操作  
    18. 
  
  
  
  
  18.     }  
    19. 
  
  
  
  
  19.   }  
    20. 
  
  
  
  
  20. } 
    21.

1、当前槽位如果头节点为null时，直接CAS写入

有人也许会质疑，如果写入时resize操作已完成，发生了table向nextTable的转变，是否会存在写入的是旧表的bin导致数据丢失的可能 ? 

这种可能性是不存在的，因为一个table在resize完成后所有的BIN都会被打上ForwardingNode的标记，可以形象的理解为所有槽位上都插满了红旗，而此处在CAS时的compare的变量null，能够保证至少在CAS原子操作发生的时间点table并未发生变更。

2、当前槽位如果头节点不为null

这里采用了一个小技巧：先锁住I槽位上的头节点，进入同步代码块后，再doubleCheck看此槽位上的头节点是否有变化。

进入同步块后还需要doubleCheck的原因：虽然一开始获取到的头节点f并非ForwardingNode，但在获取到f的同步锁之前，可能有其它线程提前获取了f的同步锁并完成了transfer工作，并将I槽位上的头节点标记为ForwardingNode，此时的f就成了一个过时的bin的头节点。

然而因为标记操作与transfer作为一个整体在同步的代码块中执行，如果doubleCheck的结果是此槽位上的头节点还是f，则表明至少在当前时间点该槽位还没有被transfer到新表(假如当前有transfer in progress的话)，可以放心的对该bin进行put/remove/replace等写操作。

只要未发生transfer或者treeify操作，链表的新增操作都是采取后入式，头节点一旦确定不会轻易改变，这种后入式的更新方式保证了锁定头节点就等于锁住了整个bin。

如果不作doubleCheck判断，则有可能当前槽位已被transfer，写入的还是旧表的BIN，从而导致写入数据的丢失；也有可能在获取到f的同步锁之前，其它线程对该BIN做了treeify操作，并将头节点替换成了TreeBin, 导致写入的是旧的链表，而非新的红黑树；

3、doubleCheck是否有ABA问题

也许有人会质疑，如果有其它线程提前对当前bin进行了的remove/put的操作，引入了新的头节点，并且恰好发生了JVM的内存释放和重新分配，导致新的Node的引用地址恰好跟旧的相同，也就是存在所谓的ABA问题。

这个可以通过反证法来推翻，在带有GC机制的语言环境下通常不会发生ABA问题，因为当前线程包含了对头节点f的引用，当前线程并未消亡，不可能存在f节点的内存被GC回收的可能性。

还有人会质疑，如果在写入过程中主哈希表发生了变化，是否可能写入的是旧表的bin导致数据丢失，这个也可以通过反证法来推翻，因为table向nextTable的转化(也就是将resize后的新哈希表正式commit)只有在所有的槽位都已经transfer成功后才会进行，只要有一个bin未transfer成功，则说明当前的table未发生变化，在当前的时间点可以放心的向table的bin内写入数据。

4、如何操作才安全

可以总结出规律，在对table的槽位成功进行了CAS操作且compare值为null，或者对槽位的非forwardingNode的头节点加锁后，doubleCheck头节点未发生变化，对bin的写操作都是安全的。

七、原子计算相关方法

原子计算主要包括：computeIfAbsent、computeIfPresent、compute、merge四个方法。 

1、几个方法的比较 

主要区别如下：

（1）computeIfAbsent只会在判断到key不存在时才会插入，判空与插入是一个原子操作，提供的FunctionalInterface是一个二元的Function, 接受key参数，返回value结果；如果计算结果为null则不做插入。

（2）computeIfPresent只会在判读单到Key非空时才会做更新，判断非空与插入是一个原子操作，提供的FunctionalInterface是一个三元的BiFunction,接受key,value两个参数，返回新的value结果；如果新的value为null则删除key对应节点。

（3）compute则不加key是否存在的限制，提供的FunctionalInterface是一个三元的BiFunction,接受key,value两个参数，返回新的value结果；如果旧的value不存在则以null替代进行计算；如果新的value为null则保证key对应节点不会存在。

（4）merge不加key是否存在的限制，提供的FunctionalInterface是一个三元的BiFunction,接受oldValue, newVALUE两个参数，返回merge后的value；如果旧的value不存在，直接以newVALUE作为最终结果，存在则返回merge后的结果；如果最终结果为null，则保证key对应节点不会存在。

2、何时会使用ReserveNode占位

如果目标bin的头节点为null，需要写入的话有两种手段：一种是生成好新的节点r后使用casTabAt(tab, i, null, r)原子操作，因为compare的值为null可以保证并发的安全；

另外一种方式是创建一个占位的ReserveNode，锁住该节点并将其CAS设置到bin的头节点，再进行进一步的原子计算操作；这两种办法都有可能在CAS的时候失败，需要自旋反复尝试。

（1）为什么只有computeIfAbsent/compute方法使用占位符的方式

computeIfPresent只有在BIN结构非空的情况下才会展开原子计算，自然不存在需要ReserveNode占位的情况；锁住已有的头节点即可。

computeIfAbsent/compute方法在BIN结构为空时，需要展开Function或者BiFunction的运算，这个操作是外部引入的需要耗时多久无法准确评估；这种情况下如果采用先计算，再casTabAt(tab, i, null, r)的方式，如果有其它线程提前更新了这个BIN，那么就需要重新锁定新加入的头节点，并重复一次原子计算(C13Map无法帮你缓存上次计算的结果，因为计算的入参有可能会变化)，这个开销是比较大的。

而使用ReserveNode占位的方式无需等到原子计算出结果，可以第一时间先抢占BIN的所有权，使其他并发的写线程阻塞。

（2）merge方法为何不需要占位

原因是如果BIN结构为空时，根据merge的处理策略，老的value为空则直接使用新的value替代，这样就省去了BiFunction中新老value进行merge的计算，这个消耗几乎是没有的；因此可以使用casTabAt(tab, i, null, r)的方式直接修改，避免了使用ReserveNode占位，锁定该占位ReserveNode后再进行CAS修改的两次CAS无谓的开销。

C13Map的compute方法 

 
 
 
 
    
  
  
  
  
  1. public V compute(K key,  
    2. 
  
  
  
  
  2.                  BiFunction remappingFunction) {  
    3. 
  
  
  
  
  3.     if (key == null || remappingFunction == null)  
    4. 
  
  
  
  
  4.         throw new nullPointerException();  
    5. 
  
  
  
  
  5.     int h = spread(key.hashCode());  
    6. 
  
  
  
  
  6.     V val = null;  
    7. 
  
  
  
  
  7.     int delta = 0;  
    8. 
  
  
  
  
  8.     int binCount = 0;  
    9. 
  
  
  
  
  9.     for (Node[] tab = table; ; ) {  
    10. 
  
  
  
  
  10.         Node f;  
    11. 
  
  
  
  
  11.         int n, i, fh;  
    12. 
  
  
  
  
  12.         if (tab == null || (n = tab.length) == 0)  
    13. 
  
  
  
  
  13.             tab = initTable();  
    14. 
  
  
  
  
  14.         else if ((f = tabAt(tab, i = (n - 1) & h)) == null) {  
    15. 
  
  
  
  
  15.             //创建占位Node  
    16. 
  
  
  
  
  16.             Node r = new ReservationNode();  
    17. 
  
  
  
  
  17.            //先锁定该占位Node  
    18. 
  
  
  
  
  18.             synchronized (r) {  
    19. 
  
  
  
  
  19.                 //将其设置到BIN的头节点  
    20. 
  
  
  
  
  20.                 if (casTabAt(tab, i, null, r)) {  
    21. 
  
  
  
  
  21.                     binCount = 1; 
    22. 
  
  
  
  
  22.                      Node node = null;  
    23. 
  
  
  
  
  23.                     try {  
    24. 
  
  
  
  
  24.                         //开始原子计算  
    25. 
  
  
  
  
  25.                         if ((val = remappingFunction.apply(key, null)) != null) {  
    26. 
  
  
  
  
  26.                             delta = 1; 
    27. 
  
  
  
  
  27.                              node = new Node(h, key, val, null);  
    28. 
  
  
  
  
  28.                         }  
    29. 
  
  
  
  
  29.                     } finally {  
    30. 
  
  
  
  
  30.                         //设置计算后的最终节点  
    31. 
  
  
  
  
  31.                         setTabAt(tab, i, node);  
    32. 
  
  
  
  
  32.                     }  
    33. 
  
  
  
  
  33.                 }  
    34. 
  
  
  
  
  34.             } 
    35. 
  
  
  
  
  35.              if (binCount != 0)  
    36. 
  
  
  
  
  36.                 break;  
    37. 
  
  
  
  
  37.         } else if ((ffh = f.hash) == MOVED)             
    
                本文标题：JavaConcurrentHashMap高并发安全实现原理解析
                
    
                文章转载：http://www.csdahua.cn/qtweb/news21/468821.html
            
    
            
    
                网站建设、网络推广公司-快上网，是专注品牌与效果的网站制作，网络营销seo公司；服务项目有等
                                
    
            
    
                广告
                
            
    
            
    
                声明：本网站发布的内容（图片、视频和文字）以用户投稿、用户转载内容为主，如果涉及侵权请尽快告知，我们将会在第一时间删除。文章观点不代表本网站立场，如需处理请联系客服。电话：028-86922220；邮箱：631063699@qq.com。内容未经允许不得转载，或转载时需注明来源：
                快上网

成都快上网为您推荐相关内容

网站收录知识

同城分类信息

重庆APP开发　　 　成都网站设计　　 　宣威网站建设　　 　八点广告　　 　成都活动策划　　 　工商财税　　 　四川成都网站制作　　 　手机网站　　 　微信服务号订阅号开发　　 　温江网站建设　　 　成都广告安装公司　　 　成都标识标牌设计制作　　 　达州服务器托管　　 　成都主机托管　　 　温江做网站　　 　免备案空间腾讯云　　 　成都网站改版　　 　全网营销　　 　官网SEO排名　　 　定制网站