【java】ConcurrentHashMap遍历 --- 弱一致性的迭代器（Iterator）实现原理

转x33g5p2x  于2022-03-21 转载在 Java  
字(8.4k)|赞(0)|评价(0)|浏览(474)

1.概述

转载：ConcurrentHashMap遍历 — 弱一致性的迭代器（Iterator）实现原理

在 Java并发编程实战（进阶篇） 中分析了 Vector 在迭代过程中对容器进行修改会抛出 ConcurrentModificationException 异常，但在并发容器中还会出现这种情况吗？

在并发容器中并不会出现这种情况，这是因为，util包中的迭代器实现是fast-failed迭代器（就是一旦由修改就抛异常），而在current包中迭代器是弱一致性迭代器。

那么我们通过 ConcurrentHashMap 源码分析一下这种弱一致性迭代器的工作原理：

测试代码在主线程向容器中预先添加一些元素，然后另外启动一个线程不断向容器中添加元素，同时在主线程使用迭代器遍历容器。此代码运行的结果是，不会抛出 ConcurrentModificationException ，同时添加到hashMap中的部分元素也会被迭代器迭代输出。

1.1 案例1

测试代码：

  1. /**
  2.      * 测试点：测试使用 ConcurrentHashMap 的时候，一边添加数据 一边遍历数据
  3.      *       添加的数据 能被遍历到吗
  4.      *
  5.      * 测试结果
  6.      *
  7.      * 不会抛出 ConcurrentModificationException ，同时添加到hashMap中的部分元素也会被迭代器迭代输出。
  8.      * 但是是乱序的
  9.      *
  10.      *
  11.      */
  12.     public void concurrentAddAndItera() {
  13.         ConcurrentHashMap<String, String> hashMap =
  14.                 new ConcurrentHashMap<>();
  16.         for (int i = 0; i < 100; i++) {
  17.             hashMap.put("" + i, "" + i);
  18.         }
  20.         Thread thread = new Thread(() -> {
  21.             for (int i = 200; i < 10000000; i++) {
  22.                 hashMap.put("" + i, "" + i);
  23.                 try {
  24.                     Thread.sleep(50);
  25.                 } catch (InterruptedException e) {
  26.                     e.printStackTrace();
  27.                 }
  28.             }
  29.         });
  30.         thread.start();
  32.         Iterator<String> iterator = hashMap.keySet().iterator();
  33.         while (iterator.hasNext()) {
  34.             String key = (String) iterator.next();
  35.             System.out.println(key);
  36.             try {
  37.                 Thread.sleep(50);
  38.             } catch (InterruptedException e) {
  39.                 e.printStackTrace();
  40.             }
  41.         }
  42.     }

1.2 案例2

可以看到确实不是全部遍历出来了。

  1. /**
  2.      * 测试点：测试使用 ConcurrentHashMap 的时候，一边添加数据 一边遍历数据
  3.      *       添加的数据 能被遍历到吗
  4.      *
  5.      *       测试能否遍历出来，所有的新加的数据
  6.      *
  7.      * 测试结果
  8.      *
  9.      * 线程退出
  10.      * 没有了
  11.      * 最终剩余：7740
  12.      * 最终剩余：2261
  13.      * 
  14.      * 可以看到有很多数据是没有遍历出来的
  15.      */
  16.     public void concurrentAddAndItera1() throws Exception,InterruptedException {
  17.         // 用于检测的hashMap
  18.         ConcurrentHashMap<String, String> checkHashMap = new ConcurrentHashMap<>();
  19.         for (int i = 0; i <= 10000; i++) {  checkHashMap.put("" + i, "" + i);  }
  21.         ConcurrentHashMap<String, String> hashMap = new ConcurrentHashMap<>();
  22.         for (int i = 0; i < 100; i++) {  hashMap.put("" + i, "" + i);  }
  24.         Thread thread = new Thread(() -> {
  25.             for (int i = 101; i <= 9000; i++) {
  26.                 hashMap.put("" + i, "" + i);
  27.                 sleep(10);
  28.             }
  29.             System.out.println("线程退出");
  30.         });
  31.         thread.start();
  33.         int count = 0;
  34.         Iterator<String> iterator = hashMap.keySet().iterator();
  35.         while (iterator.hasNext()) {
  36.             String key = (String) iterator.next();
  37. //            System.out.println(key);
  38.             checkHashMap.remove(key);
  39.             sleep(50);
  40.             count++;
  41.             if(!iterator.hasNext()) System.out.println("没有了");
  42.         }
  44.         System.out.println("最终剩余："+checkHashMap.size());
  45.         System.out.println("最终剩余："+count);
  46.     }
  48.     private void sleep(int i) {
  49.         try {
  50.             Thread.sleep(i);
  51.         } catch (InterruptedException e) {
  52.             e.printStackTrace();
  53.         }
  54.     }

2.源码分析

出现这种情况也是我们希望并发容器出现的结果，那么查看源码分析：

2.1 keySet

调用 hashMap.keySet() ，返回了一个KeySetView内部类的实例

  1. public KeySetView<K,V> keySet() {
  2.     KeySetView<K,V> ks;
  3.     return (ks = keySet) != null ? ks : (keySet = new KeySetView<K,V>(this, null));
  4. }

keySet 是 ConcurrentHashMap 类中的私有变量，存储容器中的键的 Set 集合

2.2 KeySetView

KeySetView 静态内部类，重点分析 iterator 方法，返回 KeyIterator 迭代器实例

其实简单理解，KeySetView 相当于存放 ConcurrentHashMap 键的一个 Set 集合

  1. public static class KeySetView<K,V> extends CollectionView<K,V,K>
  2.         implements Set<K>, java.io.Serializable {
  3.     private static final long serialVersionUID = 7249069246763182397L;
  4.     private final V value;
  6.     KeySetView(ConcurrentHashMap<K,V> map, V value) {  // non-public
  7.         super(map);
  8.         this.value = value;
  9.     }
  10.     
  11.     public Iterator<K> iterator() {
  12.         Node<K,V>[] t;
  13.         ConcurrentHashMap<K,V> m = map;
  14.         int f = (t = m.table) == null ? 0 : t.length;
  15.         return new KeyIterator<K,V>(t, f, 0, f, m);
  16.     }
  18.     public Spliterator<K> spliterator() {
  19.         Node<K,V>[] t;
  20.         ConcurrentHashMap<K,V> m = map;
  21.         long n = m.sumCount();
  22.         int f = (t = m.table) == null ? 0 : t.length;
  23.         return new KeySpliterator<K,V>(t, f, 0, f, n < 0L ? 0L : n);
  24.     }
  25. }

map 是 CollectionView 类中成员变量，在调用 hashMap.keySet() 方法时，会将当前 ConcurrentHashMap 传入构造方法参数中（代码见上）。

2.3 KeyIterator

KeyIterator 类继承 BaseIterator 类，而 BaseIterator 类继承 Traverser 类

下面贴出 KeyIterator 类、BaseIterator 类、 Traverser 类代码，可以看出重要实现都在 Traverser 类中，后面主要关注 Traverser 类。

KeyIterator 类：

  1. static final class KeyIterator<K,V> extends BaseIterator<K,V>
  2.         implements Iterator<K>, Enumeration<K> {
  3.         KeyIterator(Node<K,V>[] tab, int index, int size, int limit,
  4.                     ConcurrentHashMap<K,V> map) {
  5.             super(tab, index, size, limit, map);
  6.         }
  8.         public final K next() {
  9.             Node<K,V> p;
  10.             if ((p = next) == null)
  11.                 throw new NoSuchElementException();
  12.             K k = p.key;
  13.             lastReturned = p;
  14.             advance();
  15.             return k;
  16.         }
  18.         public final K nextElement() { return next(); }
  19.     }

KeyIterator 构造方法中调用 super(tab, index, size, limit, map) 父类 BaseIterator 构造方法。

2.4 BaseIterator

  1. static class BaseIterator<K,V> extends Traverser<K,V> {
  2.         final ConcurrentHashMap<K,V> map;
  3.         Node<K,V> lastReturned;
  4.         BaseIterator(Node<K,V>[] tab, int size, int index, int limit,
  5.                     ConcurrentHashMap<K,V> map) {
  6.             super(tab, size, index, limit);
  7.             this.map = map;
  8.             advance();
  9.         }
  11.         public final boolean hasNext() { return next != null; }
  12.         public final boolean hasMoreElements() { return next != null; }
  14.         public final void remove() {
  15.             Node<K,V> p;
  16.             if ((p = lastReturned) == null)
  17.                 throw new IllegalStateException();
  18.             lastReturned = null;
  19.             map.replaceNode(p.key, null, null);
  20.         }
  21.     }

BaseIterator 构造方法还是调用父类 Traverser 的构造方法，并且调用父类 advance 方法，先获取下一个遍历的有效结点。

2.5 Traverser

  1. static class Traverser<K,V> {
  2.       Node<K,V>[] tab;        // current table; updated if resized
  3.       // 下一个要访问的entry
  4.       Node<K,V> next;         // the next entry to use
  5.       // 发现forwardingNode时，stack保存当前栈顶 table 相关信息
  6.       // spare 按照 stack 的出栈顺序保存table，栈顶保存最后一个出栈的table
  7.       TableStack<K,V> stack, spare; // to save/restore on ForwardingNodes
  8.       // 下一个要访问的hash桶索引
  9.       int index;              // index of bin to use next
  10.       // 当前正在访问的初始table的hash桶索引
  11.       int baseIndex;          // current index of initial table
  12.       // 初始table的hash桶索引边界
  13.       int baseLimit;          // index bound for initial table
  14.       // 初始table的长度
  15.       final int baseSize;     // initial table size
  16.       
  17.       /**
  18.        * 如果有可能，返回下一个有效节点，否则返回null。
  19.        */
  20.       final Node<K,V> advance() {
  21.           Node<K,V> e;
  22.           //获取Node链表的下一个元素e
  23.           if ((e = next) != null)
  24.               e = e.next;
  25.           for (;;) {
  26.               Node<K,V>[] t; int i, n;  // must use locals in checks
  27.               // e不为空，返回e
  28.               if (e != null)
  29.                   return next = e;
  30.                   
  31.               // e为空，说明此链表已经遍历完成，准备遍历下一个hash桶
  32.               if (baseIndex >= baseLimit || (t = tab) == null ||
  33.                   (n = t.length) <= (i = index) || i < 0)
  34.                   // table 遍历完成，到达边界，返回null
  35.                   return next = null;
  36.                   
  37.               // e = tabAt(t, i) 获取下一个hash桶对应的node链表的头节点
  38.               if ((e = tabAt(t, i)) != null && e.hash < 0) {
  39.                   // 扩容节点,说明此hash桶中的节点已经迁移到了nextTable
  40.                   if (e instanceof ForwardingNode) {
  41.                   	  // 当前遍历 table 换做 nextTable
  42.                       tab = ((ForwardingNode<K,V>)e).nextTable;
  43.                       e = null;
  44.                       // 保存当前table的遍历状态
  45.                       pushState(t, i, n);
  46.                       continue;
  47.                   }
  48.                   //红黑树
  49.                   else if (e instanceof TreeBin)
  50.                       e = ((TreeBin<K,V>)e).first;
  51.                   else
  52.                       e = null;
  53.               }
  54.               
  55.               if (stack != null)
  56.                   // 此时遍历的是迁移目标nextTable
  57.                   // 尝试回退到源table，继续遍历源table中的节点
  58.                   recoverState(n);
  59.               else if ((index = i + baseSize) >= n)
  60.                   // 初始table的hash桶索引+1 ，即遍历下一个hash桶
  61.                   index = ++baseIndex; // visit upper slots if present
  62.           }
  63.       }
  64.       
  65.       
  66.       /**
  67.        * 在遇到扩容节点时保存遍历状态。
  68.        */
  69.       private void pushState(Node<K,V>[] t, int i, int n) {
  70.       	  // s 指向上一个出栈的 table 状态
  71.           TableStack<K,V> s = spare;  // reuse if possible
  72.           if (s != null)
  73.           	  // 如果存在上一个出栈 table 的引用
  74.           	  // 那么spare指向上上一个出栈 table
  75.               spare = s.next;
  76.           else
  77.               s = new TableStack<K,V>();
  78.           // 保存当前 table 的遍历状态    
  79.           s.tab = t;
  80.           s.length = n;
  81.           s.index = i;
  82.           s.next = stack;
  83.           stack = s;
  84.       }
  86.       /**
  87.        * 可能会弹出遍历状态。
  88.        * @param n length of current table
  89.        */
  90.       private void recoverState(int n) {
  91.           TableStack<K,V> s; int len;
  92.           /**
  93.             (s = stack) != null :
  94.             	stack不空，说明此时遍历的是nextTable  
  95.             (index += (len = s.length)) >= n : 
  96.             	确保了按照index,index+table.length的顺序遍历nextTable，
  97.             	条件成立表示nextTable已经遍历完毕
  98.           */
  99.           
  100.           // 如果进入循环，nextTable中的桶遍历完毕
  101.           while ((s = stack) != null 
  102.           	&& (index += (len = s.length)) >= n) {
  103.               // 弹出table，获取table的遍历状态，开始遍历table中的桶
  104.               n = len;
  105.               index = s.index;
  106.               tab = s.tab;
  107.               s.tab = null;
  108.               // 下面是简单的链栈操作
  109.               // 弹出 table，存入到 spare中
  110.               TableStack<K,V> next = s.next;
  111.               s.next = spare; // save for reuse
  112.               stack = next;
  113.               spare = s;
  114.           }
  115.           // 和 advance 方法最后一段代码作用相似
  116.           // table的index桶链表遍历完，遍历下一个 hash桶
  117.           // index记录下一次访问索引（当前访问索引 baseIndex++）
  118.           if (s == null && (index += baseSize) >= n)
  119.               index = ++baseIndex;
  120.       }
  121. }

在 Traverser 类中，最重要的就是 advance 方法，那么了解一下这个方法的整体思路（其中的 table 表示源table，nextTable 表示扩容之后的 table）：

当迭代器创建或者调用 next 方法是，都会调用 Traverser 中的 advance 方法来获取下一个有效结点。这个方法会根据当前 table 中 index 桶的状态来返回有效结点，如果桶是处于正常状态，那么就返回桶中的结点；如果 index 桶处于扩容状态，那么就根据 nextTable 中的 index 和 index+table.length 桶的状态来返回有效结点。在 nextTable 中访问完目标结点之后，返回到 table 再次向 ++index 桶的状态进行判断并返回，直到 table 中的桶被遍历完。

那么知道了大致的实现过程，下面使用图解的方式再次理解，加深印象：

1.当遍历到 pwd 结点的时候，说明正在扩容，此结点的数据已经被迁移到 nextTable

2.将 table1 的遍历状态压栈stack，遍历迁移到 nextTable（table2）的 hash 桶

  • 由于扩容时，会将 table(table1) 中索引为 index 桶，迁移到 nextTable(table2) 中索引为index和索引为 index+table.length 的2个桶中。因此在 nextTable(table2) 中的遍历顺序为：index，index+table.length （table2中的桶2，和桶4 )
  • 如果 桶2 和 桶4 都是正常的节点，遍历完 桶2 和 桶4 后，就会将 stack 中的table(tab1) 弹出，继续遍历 table(tab1) 中的桶
  • 桶2 是 fwd 节点，说明 table2 也被扩容，此节点的 node 被迁移到了 nextTable(table3) 中

3.将 table2 的遍历状态压栈stack，遍历迁移到 nextTable（table3）的 hash 桶

因为在 table2 中 桶2 中只有 A。根据扩容规则：在 table3 中，A 就可能在 桶2 中，也可能在 桶6 （2+table2.length）中

根据扩容迁移规则：先遍历 index = 2 的 hash 桶

接着遍历 index = 6 的 hash 桶（为什么要遍历 桶6 可见源码 recoverState 方法）

4.遍历完 table3 后，table2 出栈，根据遍历状态信息，继续遍历 table2

table2 弹出栈stack，进入了栈 spare（对象重利用，减少内存开销）
table2 记录当前 index = 2，那么接着遍历下一个 index=4（2+table1.length） 的 hash 桶
桶4 是一个 pws 结点，那么 table2 压栈stack（index = 4），直接复用 spare 指向的 table2 引用

5.遍历 nextTable（table3）中的 桶4、桶8（4+table2.length）

6.完成 nextTable（table3）的遍历，table2 出栈并遍历

table2 遍历完毕。

7.table1 出栈并遍历

stack 栈空，而且 spare 栈按照出栈的顺序保存了相关节点，遍历完毕。

至此 table 的图解遍历就完成了。

在 advance 方法中通过不断循环遍历，其中考虑到table的大小发生变化，并且节点的组织方式可能是链表也可能是红黑树，遍历的过程中可能会有部分数据遍历不到，此为弱一致性的表现。

版权说明 : 本文为转载文章, 版权归原作者所有 版权申明

原文链接 : https://blog.csdn.net/qq_21383435/article/details/123618422

内容来源于网络，如有侵权，请联系作者删除！

JavaMap遍历弱一致性

相关文章

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新文章

    更多
    • 蜀ICP备13028337号-1 大数据知识库 https://www.saoniuhuo.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com