ConcurrentHashMap高并发实现原理

文章目录 隐藏
1 ConcurrentHashMap结构
2 重要属性
3 重要内部类
4 三个核心方法
5 初始化initTable
6 扩容
7 put方法
8 get方法

ConcurrentHashMap结构

在JDK7中ConcurrentHashMap是基于细粒度分离锁实现的，其结果大致如下：

在JDK7中ConcurrentHashMap由多个Segment组成，每个Segment都继承了ReentrantLock，是一把重入锁。

JDK8 中完全重写了ConcurrentHashmap，ConcurrentHashmap实现上和原来的分段式存储有很大的区别，本文讲解的JDK8中ConcurrentHashMap与JDK8的HashMap有相通之处，底层依然由“数组”+链表+红黑树实现，利用全新的CAS算法保证线程安全。Segment虽保留，但已经简化属性，仅仅是为了兼容旧版本。

CAS算法：unsafe.compareAndSwapInt(this, valueOffset, expect, update);  CAS(Compare And Swap)，意思是如果valueOffset位置包含的值与expect值相同，则更新valueOffset位置的值为update，并返回true，否则不更新，返回false。

CAS作为知名无锁算法，那ConcurrentHashMap就没用锁了么？当然不是，hash值相同的链表的头结点还是会synchronized上锁。

重要属性

我们重点关注几个出镜率比较高的属性

    private transient volatile int sizeCtl;

 sizeCtl是控制标识符，volatile关键字保证了其线程可见性，不同的值表示不同的意义。

负数代表正在进行初始化或扩容操作
-1代表正在初始化
-N 表示有N-1个线程正在进行扩容操作
正数或0代表hash表还没有被初始化，这个数值表示初始化或下一次进行扩容的大小，类似于扩容阈值。它的值始终是当前ConcurrentHashMap容量的0.75倍，这与loadfactor是对应的。实际容量>=sizeCtl，则扩容。

    transient volatile Node<K,V>[] table;

盛装Node元素的数组，在第一次插入时被初始化（延迟初始化，并没有在构造函数执行时进行初始化），它的大小总是2的整数次幂 。

//链表转树阈值，大于8时    
static final int TREEIFY_THRESHOLD = 8;
//树转链表阈值，小于等于6时（仅在扩容transfer时才可能树转链表）
static final int UNTREEIFY_THRESHOLD = 6;

重要内部类

Node是最核心的内部类，它包装了key-value键值对，所有插入ConcurrentHashMap的数据都包装在这里面。

相关注意的点在源码中用注释说明：

  
static class Node
    
      implements Map.Entry
     
       { final int hash;//final 修饰 final K key; //final 修饰 volatile V val; // volatile 保证线程可见性 volatile Node
      
        next; //volatile保证线程可见性 
       
    Node(int hash, K key, V val, Node<K,V> next) {
        this.hash = hash;
        this.key = key;
        this.val = val;
        this.next = next;
    }

    public final K getKey()       { return key; }
    public final V getValue()     { return val; }
    public final int hashCode()   { return key.hashCode() ^ val.hashCode(); }
    public final String toString(){ return key + "=" + val; }
   //不允许直接改变value的值
    public final V setValue(V value) {
        throw new UnsupportedOperationException();
    }
   //增加find方法辅助get
    public final boolean equals(Object o) {
        Object k, v, u; Map.Entry<?,?> e;
        return ((o instanceof Map.Entry) &amp;&amp;
                (k = (e = (Map.Entry<?,?>)o).getKey()) != null &amp;&amp;
                (v = e.getValue()) != null &amp;&amp;
                (k == key || k.equals(key)) &amp;&amp;
                (v == (u = val) || v.equals(u)));
    }



      
,v>

     ,v>

    ,v>




TreeNode是另外一个核心的数据结构。当链表长度过长的时候，会转换为TreeNode。但是与HashMap不相同的是，它并不是直接转换为红黑树，而是把这些结点包装成TreeNode放在TreeBin对象中，由TreeBin完成对红黑树的包装。



    /**
     * Nodes for use in TreeBins 链表>8，才可能转为TreeNode
     */ 
    static final class TreeNode
    
      extends Node
     
       { TreeNode
      
        parent; // red-black tree links TreeNode
       
         left; TreeNode
        
          right; TreeNode
         
           prev; // needed to unlink next upon deletion boolean red; 
          
    TreeNode(int hash, K key, V val, Node<K,V> next,
             TreeNode<K,V> parent) {
        super(hash, key, val, next);
        this.parent = parent;
    }

    Node<K,V> find(int h, Object k) {
        return findTreeNode(h, k, null);
    }

    /**
     * Returns the TreeNode (or null if not found) for the given key
     * starting at given root.
     */
    final TreeNode<K,V> findTreeNode(int h, Object k, Class<?> kc) {
        if (k != null) {
            TreeNode<K,V> p = this;
            do  {
                int ph, dir; K pk; TreeNode<K,V> q;
                TreeNode<K,V> pl = p.left, pr = p.right;
                if ((ph = p.hash) > h)
                    p = pl;
                else if (ph < h)
                    p = pr;
                else if ((pk = p.key) == k || (pk != null &amp;&amp; k.equals(pk)))
                    return p;
                else if (pl == null)
                    p = pr;
                else if (pr == null)
                    p = pl;
                else if ((kc != null ||
                          (kc = comparableClassFor(k)) != null)
                         (dir = compareComparables(kc, k, pk)) != 0)
                    p = (dir < 0) ? pl : pr;
                else if ((q = pr.findTreeNode(h, k, kc)) != null)
                    return q;
                else
                    p = pl;
            } while (p != null);
        }
        return null;
    }
}



         
,v>

        ,v>

       ,v>

      ,v>

     ,v>

    ,v>




TreeNode继承自ConcurrentHashMap的Node类而并非HashMap中的继承自LinkedHashMap.Entry

   

    类，也就是说TreeNode带有next指针，这样做的目的是方便基于TreeBin的访问。

   ,v>


TreeBin并不负责包装用户的key、value信息，而是包装的很多TreeNode节点。它代替了TreeNode的根节点，也就是说在实际的ConcurrentHashMap“数组”中，存放的是TreeBin对象，而不是TreeNode对象，这是与HashMap的区别。另外这个类还带有了读写锁。


    
static final class TreeBin
    
      extends Node
     
       { TreeNode
      
        root; volatile TreeNode
       
         first; volatile Thread waiter; volatile int lockState; // values for lockState static final int WRITER = 1; // set while holding write lock static final int WAITER = 2; // set when waiting for write lock static final int READER = 4; // increment value for setting read lock ...... /** * Creates bin with initial set of nodes headed by b. * 构造红黑树 */ TreeBin(TreeNode
        
          b) { super(TREEBIN, null, null, null); this.first = b; TreeNode
         
           r = null; for (TreeNode
          
            x = b, next; x != null; x = next) { next = (TreeNode
           
            )x.next; x.left = x.right = null; if (r == null) { x.parent = null; x.red = false; r = x; } else { K k = x.key; int h = x.hash; Class
             kc = null; for (TreeNode
            
              p = r;;) { int dir, ph; K pk = p.key; if ((ph = p.hash) > h) dir = -1; else if (ph < h) dir = 1; else if ((kc == null && (kc = comparableClassFor(k)) == null) || (dir = compareComparables(kc, k, pk)) == 0) dir = tieBreakOrder(k, pk); TreeNode
             
               xp = p; if ((p = (dir <= 0) ? p.left : p.right) == null) { x.parent = xp; if (dir <= 0) xp.left = x; else xp.right = x; r = balanceInsertion(r, x); break; } } } } this.root = r; assert checkInvariants(root); } ...... } 
             ,v>
            ,v>
           ,v>
          ,v>
         ,v>
        ,v>
       ,v>
      ,v>
     ,v>
    ,v>


ForwardingNode一个用于连接两个table的节点类，它包含一个nextTable指针，用于指向下一张表。而且这个节点的key value next指针全部为null，它的hash值为-1。生命周期：仅存活于扩容操作且bin不为null时，一定会出现在每个bin的首位。


  
/**
     * A node inserted at head of bins during transfer operations.
     */
    static final class ForwardingNode
    
      extends Node
     
       { final Node
      
       [] nextTable; ForwardingNode(Node
       
        [] tab) { super(MOVED, null, null, null); this.nextTable = tab; } //从nextTable里进行查询节点，而不是以自身为头结点进行查找 Node
        
          find(int h, Object k) { // loop to avoid arbitrarily deep recursion on forwarding nodes outer: for (Node
         
          [] tab = nextTable;;) { Node
          
            e; int n; if (k == null || tab == null || (n = tab.length) == 0 || (e = tabAt(tab, (n - 1) & h)) == null) return null; for (;;) { int eh; K ek; if ((eh = e.hash) == h && ((ek = e.key) == k || (ek != null && k.equals(ek)))) return e; if (eh < 0) { if (e instanceof ForwardingNode) { tab = ((ForwardingNode
           
            )e).nextTable; continue outer; } else return e.find(h, k); } if ((e = e.next) == null) return null; } } } } 
           ,v>
          ,v>
         ,v>
        ,v>
       ,v>
      ,v>
     ,v>
    ,v>


三个核心方法


ConcurrentHashMap定义了三个原子操作，用于对指定位置的节点进行操作。正是这些原子操作保证了ConcurrentHashMap的线程安全。



//获取i位置上的Node节点  
static final 
    
      Node
     
       tabAt(Node
      
       [] tab, int i) { return (Node
       
        )U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE); } //利用CAS算法设置i位置上的结点Node,之所以能实现并发是因为它指定了原来这个节点的值是多少 //在CAS算法中会比较内存中的值与你指定的值是否相等，如果相等才接受你的修改，否则拒绝你的修改 //因为当前线程中的值并不是最新的值，这种修改可能会覆盖其他线程的修改结果 static final 
        
          boolean casTabAt(Node
         
          [] tab, int i, Node
          
            c, Node
           
             v) { return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v); } //利用volatile方法设置节点位置i的值，仅在上锁区被调用 static final 
            
              void setTabAt(Node
             
              [] tab, int i, Node
              
                v) { U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v); } 
              ,v>
             ,v>
            ,v>
           ,v>
          ,v>
         ,v>
        ,v>
       ,v>
      ,v>
     ,v>
    ,v>


初始化initTable


对于ConcurrentHashMap来说，调用它的构造方法仅仅是设置了一些参数而已。而整个table的初始化是在向ConcurrentHashMap中插入第一个元素的时候发生的。


  
/**
     * Initializes table, using the size recorded in sizeCtl.
     */
    private final Node
    
     [] initTable() { Node
     
      [] tab; int sc; while ((tab = table) == null || tab.length == 0) { if ((sc = sizeCtl) < 0) Thread.yield(); // lost initialization race; just spin else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) { try { if ((tab = table) == null || tab.length == 0) { int n = (sc > 0) ? sc : DEFAULT_CAPACITY; @SuppressWarnings("unchecked") Node
      
       [] nt = (Node
       
        [])new Node
        [n]; table = tab = nt; sc = n - (n >>> 2); } } finally { sizeCtl = sc; } break; } } return tab; } 
       ,v>
      ,v>
     ,v>
    ,v>



初始化方法主要应用了关键属性sizeCtl 如果这个值小于0，表示其他线程正在进行初始化，就放弃这个操作。在这也可以看出ConcurrentHashMap的初始化只能由一个线程完成。如果获得了初始化权限，就用CAS方法将sizeCtl置为-1，防止其他线程进入。初始化数组后，将sizeCtl的值改为0.75*n。

扩容

当ConcurrentHashMap容量不足的时候，需要对table进行扩容。这个方法的基本思想跟HashMap是很像的，但是由于它是支持并发扩容的，所以要复杂的多。

原因是它支持多线程进行扩容操作，而并没有加锁。我想这样做的目的不仅仅是为了满足concurrent的要求，而是希望利用并发处理去减少扩容带来的时间影响。因为在扩容的时候，总是会涉及到从一个“数组”到另一个“数组”拷贝的操作，如果这个操作能够并发进行，那真真是极好的了。

整个扩容操作分两部分：

  1. 构建一个nextTable，它的容量是原来的两倍，这个线程是单线程操作的。单线程的保证是通过RESIZE_STAMP_SHIFT变量经过一次运算来保证的。
  2. 将原来的table数组复制到nextTable中，这里允许多线程操作。
/** The next table to use; non-null only while resizing.
     * 一个过渡的table表，只有在扩容时才使用
     */
    private transient volatile Node<K,V>[] nextTable;

我们首先看下单个线程是如何完成的，它的大致思路就是遍历、复制的过程。首先通过运算得到需要遍历的次数i,然后利用tabAt方法获取位置i处的元素。

（1）如果这个位置为空，就在原table的i位置放入ForwardingNode节点，这个也是触发并发扩容的关键点。

（2）如果这个位置是Node节点（fh>=0），如果它是一个链表的头节点，就构造一个反序链表，把它们分别放在nextTable的i和i+n位置。

（3）如果这个节点是TreeBin节点（fh<0），也做一个反序处理，并且判断是否需要untreefi,把处理结果分别放在nextTable的i和i+n位置。

（4）遍历完所有节点后就完成了复制工作。这时让nextTable作为新的table,并且更新sizeCtl为新容量的0.75倍，完成扩容。

 

再看下多线程的情况

如果线程遍历到的节点是forward节点，就向后继续遍历，再加上给节点上锁的机制，就完成了多线程的控制。多线程遍历节点，处理了一个节点，就把对应点的值set为forward，另一个线程看到forward，就向后遍历。这样交叉就完成了复制工作。而且还很好的解决了线程安全的问题。这个方法设计很值得学习，数据复制的结构图如下：

 

 
/* Moves and/or copies the nodes in each bin to new table. See
     * above for explanation.
     */
    private final void transfer(Node
   
    [] tab, Node
    
     [] nextTab) { int n = tab.length, stride; if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE) stride = MIN_TRANSFER_STRIDE; // subdivide range if (nextTab == null) { // initiating try { @SuppressWarnings("unchecked") Node
     
      [] nt = (Node
      
       [])new Node
       [n << 1];//构造一个nextTable表，它的容量是原来的两倍 nextTab = nt; } catch (Throwable ex) { // try to cope with OOME sizeCtl = Integer.MAX_VALUE; return; } nextTable = nextTab; transferIndex = n; } int nextn = nextTab.length; ForwardingNode
       
         fwd = new ForwardingNode
        
         (nextTab);//构造一个连节点指针，用于标识位 boolean advance = true;//并发扩容关键变量，如果为true说明这个节点已经处理过 boolean finishing = false; // to ensure sweep before committing nextTab for (int i = 0, bound = 0;;) { Node
         
           f; int fh; //while循环，依次遍历hash表中的节点 while (advance) { int nextIndex, nextBound; if (--i >= bound || finishing) advance = false; else if ((nextIndex = transferIndex) <= 0) { i = -1; advance = false; } else if (U.compareAndSwapInt (this, TRANSFERINDEX, nextIndex, nextBound = (nextIndex > stride ? nextIndex - stride : 0))) { bound = nextBound; i = nextIndex - 1; advance = false; } } if (i < 0 || i >= n || i + n >= nextn) { int sc; if (finishing) { //如果所有节点都完成了复制工作，就把nextTable赋值给table,并清空临时对象nextTable nextTable = null; table = nextTab; sizeCtl = (n << 1) - (n >>> 1);//扩容阈值设置为原来的1.5倍，依然相当于现在容量的0.75 return; } //CAS算法更新扩容阈值，在这里sizeCtl减1，说明新加入一个线程参与到扩容操作 if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) { if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT) return; finishing = advance = true; i = n; // recheck before commit } } //如果遍历到的节点为空，则放入ForwardingNode指针 else if ((f = tabAt(tab, i)) == null) advance = casTabAt(tab, i, null, fwd); //如果遍历大ForwardingNode节点，说明这个节点节点已经被处理过，直接跳过，这是控制并发扩容的核心 else if ((fh = f.hash) == MOVED) advance = true; // already processed else {//节点上锁 synchronized (f) { if (tabAt(tab, i) == f) { Node
          
            ln, hn; if (fh >= 0) {//fh>=0 证明是Node节点 int runBit = fh & n; //以下部分完成构建两个链表，一个是原链表，另一个是原链表的反序链表 Node
           
             lastRun = f; for (Node
            
              p = f.next; p != null; p = p.next) { int b = p.hash & n; if (b != runBit) { runBit = b; lastRun = p; } } if (runBit == 0) { ln = lastRun; hn = null; } else { hn = lastRun; ln = null; } for (Node
             
               p = f; p != lastRun; p = p.next) { int ph = p.hash; K pk = p.key; V pv = p.val; if ((ph & n) == 0) ln = new Node
              
               (ph, pk, pv, ln); else hn = new Node
               
                (ph, pk, pv, hn); } setTabAt(nextTab, i, ln);//在nextTable的i位置插入链表 setTabAt(nextTab, i + n, hn);//在nextTable的i+n位置插入链表 setTabAt(tab, i, fwd);//在table的i位置插入ForwardingNode节点，表示已经处理过该节点 advance = true;//设置true，遍历下一个节点 } else if (f instanceof TreeBin) {//对TreeBin对象进行处理，和上面相似 TreeBin
                
                  t = (TreeBin
                 
                  )f; TreeNode
                  
                    lo = null, loTail = null; TreeNode
                   
                     hi = null, hiTail = null; int lc = 0, hc = 0; //构建正反两个链表 for (Node
                    
                      e = t.first; e != null; e = e.next) { int h = e.hash; TreeNode
                     
                       p = new TreeNode
                      
                        (h, e.key, e.val, null, null); if ((h & n) == 0) { if ((p.prev = loTail) == null) lo = p; else loTail.next = p; loTail = p; ++lc; } else { if ((p.prev = hiTail) == null) hi = p; else hiTail.next = p; hiTail = p; ++hc; } } //如果扩容后不再需要Tree结构，反向转换为链表 ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) : (hc != 0) ? new TreeBin
                       
                        (lo) : t; hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) : (lc != 0) ? new TreeBin
                        
                         (hi) : t; setTabAt(nextTab, i, ln); setTabAt(nextTab, i + n, hn); setTabAt(tab, i, fwd); advance = true; } } } } } } 
                        ,v>
                       ,v>
                      ,v>
                     ,v>
                    ,v>
                   ,v>
                  ,v>
                 ,v>
                ,v>
               ,v>
              ,v>
             ,v>
            ,v>
           ,v>
          ,v>
         ,v>
        ,v>
       ,v>
      ,v>
     ,v>
    ,v>
   ,v>

put方法

前面的所有的介绍其实都为这个方法做铺垫。ConcurrentHashMap最常用的就是put和get两个方法。

现在来介绍put方法，这个put方法依然沿用HashMap的put方法的思想，根据hash值计算这个新插入的点在table中的位置i，如果i位置是空的，直接放进去，否则进行判断，如果i位置是树节点，按照树的方式插入新的节点，否则把i插入到链表的末尾。

ConcurrentHashMap中依然沿用这个思想，有一个最重要的不同点就是ConcurrentHashMap不允许key或value为null值。另外由于涉及到多线程，put方法就要复杂一点。 在多线程中可能出现下面情况：

  1. 如果一个或多个线程正在对ConcurrentHashMap进行扩容，当前线程也要进入扩容的操作中。这个扩容的操作之所以能被检测到，是因为transfer方法中在空结点上插入forward节点，如果检测到需要插入的位置被forward节点占有，就帮助进行扩容；
  2. 如果检查到要插入的节点是非空且不说ForwardingNode节点，就对这个节点进行加锁，保证线程安全。尽管这有些影响性能，但还是比HashTable的synchronized要好很多。

整个流程首先定义不允许key或value为null的情况放入，对于每个放入的值，首先利用spread方法对key的hashCode进行一次hash计算，由此决定这个值在table中的位置。

如果这个位置的值为空，不需要加锁操作，利用 CAS 尝试写入，失败则自旋保证成功。

如果这个位置存在节点，说明发生了hash碰撞，首先判断这个节点的类型，如果是链表节点（fh>=0），则得到的节点就是hash值相同的节点组成的链表的头节点。如果遇到hash值与key值都与新加入节点是一致的情况，则只需要更新value值即可。否则依次向后遍历，直到链表尾插入这个结点。如果加入这个节点以后链表长度大于8，就把这个链表转换成红黑树。如果这个节点的类型已经是树节点的话，直接调用树节点的插入方法进行插入新的值。

  
/** Implementation for put and putIfAbsent */
    final V putVal(K key, V value, boolean onlyIfAbsent) {
//不允许key或value为null
        if (key == null || value == null) throw new NullPointerException();
//计算hash值
        int hash = spread(key.hashCode());
        int binCount = 0;
//死循环 何时插入成功，何时跳出
        for (Node
   
    [] tab = table;;) { Node
    
      f; int n, i, fh; //如果table为空，初始化table if (tab == null || (n = tab.length) == 0) tab = initTable(); //根据hash值，计算出table里面的位置 else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) { if (casTabAt(tab, i, null, new Node
     
      (hash, key, value, null))) break; // no lock when adding to empty bin 如果这个位置没有值，直接放入，不需要加锁 } else if ((fh = f.hash) == MOVED)//当遇到表连接点时，需要进行整合表的操作 tab = helpTransfer(tab, f); else { V oldVal = null; synchronized (f) {//节点上锁，这个节点可以理解为hash值相同节点组成的链表头节点 if (tabAt(tab, i) == f) { if (fh >= 0) {//说明是链表节点 binCount = 1; for (Node
      
        e = f;; ++binCount) {//遍历链表所有节点 K ek; //如果hash值，key值都相同，则修改对应的value if (e.hash == hash && ((ek = e.key) == key || (ek != null && key.equals(ek)))) { oldVal = e.val; if (!onlyIfAbsent) e.val = value; break; } Node
       
         pred = e; //如果遍历到最后的节点，那么就证明新的节点需要插入，，就把它插入到链表尾部 if ((e = e.next) == null) { pred.next = new Node
        
         (hash, key, value, null); break; } } } else if (f instanceof TreeBin) {//如果节点是树节点，就按照树 的方式进行插入 Node
         
           p; binCount = 2; if ((p = ((TreeBin
          
           )f).putTreeVal(hash, key, value)) != null) { oldVal = p.val; if (!onlyIfAbsent) p.val = value; } } } } if (binCount != 0) {//如果链表节点大于8,就需要把链表结构转换为树结构 if (binCount >= TREEIFY_THRESHOLD) treeifyBin(tab, i); if (oldVal != null) return oldVal; break; } } } addCount(1L, binCount);//当前ConcurrentHashMap的元素数量加1 return null; } 
          ,v>
         ,v>
        ,v>
       ,v>
      ,v>
     ,v>
    ,v>
   ,v>

get方法

get方法比较简单，给定一个key来确定value的时候，必须满足两个条件  key相同  hash值相同，对于节点可能在链表或树上的情况，需要分别去查找。

  
public V get(Object key) {
        Node
   
    [] tab; Node
    
      e, p; int n, eh; K ek; int h = spread(key.hashCode());//计算hash值 //根据hash值确定节点位置 if ((tab = table) != null && (n = tab.length) > 0 && (e = tabAt(tab, (n - 1) & h)) != null) { if ((eh = e.hash) == h) {//如果搜索的节点key与传入的key相同且不为null,直接返回这个节点 if ((ek = e.key) == key || (ek != null && key.equals(ek))) return e.val; } else if (eh < 0)//说明节点在树上 return (p = e.find(h, key)) != null ? p.val : null; while ((e = e.next) != null) {//否则遍历链表找到对应的值 if (e.hash == h && ((ek = e.key) == key || (ek != null && key.equals(ek)))) return e.val; } } return null; }
    ,v>
   ,v>
0 0 投票数
Article Rating
订阅评论
提醒
guest
0 评论
最旧
最新 最多投票
内联反馈
查看所有评论
0
希望看到您的想法，请您发表评论x