并发编程(4)ConcurrentHashMap源码分析-白红宇

概述

ConcurrentHashMap,通过这个名字，可以知道Concurrent是并发的，HashMap是我们常用的一种用来存放键值对的数据结构，所以ConcurrentHashMap就是一种用来解决高并发的HashMap。这个是JDK在1.5之后提供的一种数据结构。我们知道Java中其实已经有了HashTable这个线程安全的Map，但是他是通过对整个table使用synchronized来保证线程安全的，当有多个线程对HashTable进行读写的时候，每次只能有一个线程会获取到HashTable的对象锁，别的只能处于等待状态，所以在高并发的情况下，它的效率是比较低的。当然JDK还提供了另外一种方式，那就是Collections.synchronizedMap(hashMap)，这个底层也是通过synchronized对要高并发的数据结构来实现线程安全的。通过上面的分析可以知道，ConcurrentHashMap为了解决synchronized在高并发时的低效问题，ConcurrentHashMap之所以比HashTable高效，是因为他并不是对整个HashMap上锁，而且在JDK1.8跟之前的JDK1.6，1.7的实现都不一样，下面进行对比一下

JDK 1.6:ConcurrentHashMap是由Segment数组结构和HashEntry数组结构组成。Segment是一种可重入锁ReentrantLock，在ConcurrentHashMap里扮演锁的角色，HashEntry则用于存储键值对数据。一个ConcurrentHashMap里包含一个Segment数组，Segment的结构和HashMap类似，是一种数组和链表结构，一个Segment里包含一个HashEntry数组，每个HashEntry是一个链表结构的元素，每个Segment守护者一个HashEntry数组里的元素,当对HashEntry数组的数据进行修改时，必须首先获得它对应的Segment锁。

JDK 1.8：放弃了Segment的概念，而是直接用Node数组+链表+红黑树的数据结构来实现，并发控制使用Synchronized和CAS来操作，整个看起来就像是优化过且线程安全的HashMap。

继承关系

通过继承关系，可以发现，ConcurrentHashMap并没有继承HashMap,而是单独实现了ConcurrentMap这个接口，其余的跟HashMap是一样的

成员变量

//table的最大容量    private static final int MAXIMUM_CAPACITY = 1 << 30;    //table的默认容量16    private static final int DEFAULT_CAPACITY = 16;	//默认的并发度为16    private static final int DEFAULT_CONCURRENCY_LEVEL = 16;    //负载因子    private static final float LOAD_FACTOR = 0.75f;	//链表转化为红黑树的阈值    static final int TREEIFY_THRESHOLD = 8;	//红黑树转链表的阈值    static final int UNTREEIFY_THRESHOLD = 6;	//红黑树的最小容量    static final int MIN_TREEIFY_CAPACITY = 64;	//存放Node的数组    transient volatile Node
    
     [] table;    private transient volatile Node
     
      [] nextTable;    private transient volatile long baseCount;    //sizeCtl作为table是否在初始化或者在resize的标志，如果为负数说明正在初始化或者扩容，需要table进行CRUD操作的线程只能等待，稍后进行重试，否则就可以直接进行CRUD操作    //正数则表明table没有初始化    private transient volatile int sizeCtl;    //该节点是否已经移动到新数组中去    static final int MOVED     = -1;     //是否是TREEBIN节点    static final int TREEBIN   = -2; 复制代码

除此之外，还有几个类需要解释一下

基本原理

TreeNode

static final class TreeNode
    
      extends Node
     
       {        TreeNode
      
        parent;  //根节点        TreeNode
       
         left;//左节点        TreeNode
        
          right;//右节点        TreeNode
         
           prev; //上个节点的指针 boolean red;//默认节点为黑 TreeNode(int hash, K key, V val, Node
          
            next, TreeNode
           
             parent) { super(hash, key, val, next); this.parent = parent; } Node
            
              find(int h, Object k) { return findTreeNode(h, k, null); } //查找相应的树节点 final TreeNode
             
               findTreeNode(int h, Object k, Class
               kc) { //此处省略一万行代码...... }复制代码

跟一般的树节点都差不多，在前面分析二叉树的时候也差不多是这种数据结构，所以还是比较好理解的，注释说Nodes for use in TreeBins，也就是说这个TreeNode是用来服务TreeBin的，那么继续看Treebins

TreeBins

这个通过注释可以知道，是用来包装TreeNode的，然后作为红黑树的树节点

static final class TreeBin
    
      extends Node
     
       {TreeNode
      
        root;  volatile TreeNode
       
         first;//TreeNode的根节点  volatile Thread waiter;//waiter线程  volatile int lockState;//当前节点的锁状态    // values for lockState        static final int WRITER = 1; // 获得写数据的锁状态        static final int WAITER = 2; // 等待写数据的锁状态        static final int READER = 4; // 添加数据时的锁状态        TreeBin(TreeNode
        
          b) {            super(TREEBIN, null, null, null);            this.first = b;//头结点赋值            TreeNode
         
           r = null; //递归遍历传过来的Node，从根节点开始 for (TreeNode
          
            x = b, next; x != null; x = next) { next = (TreeNode
           
            )x.next;//拿到下一个节点 x.left = x.right = null;//将左右节点置空，因为需要重新赋值 //判断r节点是否为空 if (r == null) { x.parent = null;//根节点置空 x.red = false;//黑子树 r = x;//x赋值为r节点 } else { K k = x.key;//拿到key int h = x.hash;//计算hash值 Class
             kc = null;//K对应的泛型实体类 //从根节点r开始遍历 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;//将r赋值给根节点 assert checkInvariants(root); } }复制代码

ForwardingNode

static final class ForwardingNode
    
      extends Node
     
       {      final Node
      
       [] nextTable;        ForwardingNode(Node
       
        [] tab) {            super(MOVED, null, null, null);            this.nextTable = tab;        }        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; } } } }复制代码

ForwardingNode有两个功能，一个是用来指向nextTable节点，便于将原有的table数组复制到新的table中，另外一个是用来占位空节点，告诉其他节点这个节点已经被处理过了，自动遍历后面的节点。

CAS操作

tabAt

static final 
    
      Node
     
       tabAt(Node
      
       [] tab, int i) {        return (Node
       
        )U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE);    }复制代码

获得在i索引位置上的Node节点

casTabAt

static final 
    
      boolean casTabAt(Node
     
      [] tab, int i, Node
      
        c, Node
       
         v) {        return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v);    }复制代码

1.利用CAS算法设置i位置上的Node节点。之所以能实现并发是因为他指定了原来这个节点的值是多少

2.在CAS算法中，会比较内存中的值与你指定的这个值是否相等，如果相等才接受你的修改，否则拒绝你的修改

setTabAt

static final 
    
      void setTabAt(Node
     
      [] tab, int i, Node
      
        v) {        U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v);    }复制代码

利用volatile方法设置节点位置的值

核心方法

构造方法

之前分析过HashMap的源码，所以看这些还是比较轻松的，ConcurrentHashMap采用的是懒加载，也就是存入第一对KeyValue的时候才会去初始化table，所以前面四个构造方法只是做了一些参数配置，只有最后一个构造方法，才会去真正的初始化，下面重点看一下最后一个方法

public ConcurrentHashMap(Map
     m) {        this.sizeCtl = DEFAULT_CAPACITY;        putAll(m);    }   public void putAll(Map
     m) {       tryPresize(m.size());        for (Map.Entry
     e : m.entrySet())            putVal(e.getKey(), e.getValue(), false);    }复制代码

最后调用了putAll方法，这个会在后面的put方法里面再重点讲解。

transfer方法

transfer实际上执行的就是table数组的扩容操作，

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")                //创建一个容量是原来2倍的table数组                Node
      
       [] nt = (Node
       
        [])new Node
        [n << 1];                nextTab = nt;            } catch (Throwable ex) {      // try to cope with OOME                sizeCtl = Integer.MAX_VALUE;                return;            }            nextTable = nextTab;            transferIndex = n;        }        int nextn = nextTab.length;        //创建一个ForwardingNode，指向新table，并且用于填充已经移动的Node位置        ForwardingNode
        
          fwd = new ForwardingNode
         
          (nextTab); boolean advance = true; boolean finishing = false; // to ensure sweep before committing nextTab for (int i = 0, bound = 0;;) { Node
          
            f; int fh; 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; // 扩容至现在容量的0.75倍 sizeCtl = (n << 1) - (n >>> 1); return; } 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 } } else if ((f = tabAt(tab, i)) == null) //用ForwardingNode来填充Null节点 advance = casTabAt(tab, i, null, fwd); else if ((fh = f.hash) == MOVED) //如果当前节点移动过，将次标志设置为true //这样后续线程访问便可以直接跳过此节点 advance = true; // already processed else { synchronized (f) { if (tabAt(tab, i) == f) { Node
           
             ln, hn; if (fh >= 0) { //普通Node节点 int runBit = fh & n;//通过runBit将链表打散，便于高效遍历 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); setTabAt(nextTab, i + n, hn); setTabAt(tab, i, fwd); advance = 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); //通过h & n的值将TreeBin的节点进行打散 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; //在nextTable的i位置上插入一个链表 setTabAt(nextTab, i, ln); //在nextTable的i+n的位置上插入另一个链表 setTabAt(nextTab, i + n, hn); //在table的i位置上插入forwardNode节点 表示已经处理过该节点 setTabAt(tab, i, fwd); //设置advance为true 返回到上面的while循环中 就可以执行i--操作 advance = true; } } } } } }复制代码

1.根据当前数组长度n，新建一个两倍长度的数组nextTable；

2.初始化ForwardingNode节点，其中保存了新数组nextTable的引用，在处理完每个槽位的节点之后当做占位节点，表示该槽位已经处理过了

3.通过for自循环处理每个槽位中的链表元素，默认advace为真，通过CAS设置transferIndex属性值

4.如果槽位中没有节点，则通过CAS插入在第二步中初始化的ForwardingNode节点，用于告诉其它线程该槽位已经处理过了

5.如果当前Node已经被线程A处理了，那么线程B处理到这个节点时，取到该节点的hash值应该为MOVED，值为-1，则直接跳过，继续处理下一个Node

6.处理不为null的节点，如果是链表结构，先定义两个变量节点ln和hn,通过CAS把ln链表设置到新数组的i位置，hn链表设置到i+n的位置

7.如果该槽位是红黑树结构，同样将节点分成两类，通过CAS插入到nextTable中

helpTransfer

如果一个线程在进行扩容的时候中断了，但是此时nextTable已经创建了，那么他就会调用helpTransfer方法，帮助扩容

final Node
    
     [] helpTransfer(Node
     
      [] tab, Node
      
        f) {        Node
       
        [] nextTab; int sc;        if (tab != null && (f instanceof ForwardingNode) &&            (nextTab = ((ForwardingNode
        
         )f).nextTable) != null) {            int rs = resizeStamp(tab.length);            while (nextTab == nextTable && table == tab &&                   (sc = sizeCtl) < 0) {                if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||                    sc == rs + MAX_RESIZERS || transferIndex <= 0)                    break;                if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {                    transfer(tab, nextTab);                    break;                }            }            return nextTab;        }        return table;    }复制代码

这是一个协助扩容的方法。这个方法被调用的时候，当前ConcurrentHashMap一定已经有了nextTable对象，首先拿到这个nextTable对象，调用上面讲到的transfer方法来进行扩容

put方法

首先调用了put方法

public V put(K key, V value) {    return putVal(key, value, false);}复制代码

紧接着调用了putVal方法，下面来分析下一年putVal方法

final V putVal(K key, V value, boolean onlyIfAbsent) {  //key和value都不能为空   if (key == null || value == null) throw new NullPointerException();        int hash = spread(key.hashCode());//计算hash值        int binCount = 0;//链表长度        //开启死循环        for (Node
    
     [] tab = table;;) {            Node
     
       f; int n, i, fh;            if (tab == null || (n = tab.length) == 0)            //table为null或者table的长度为0，初始化table                tab = initTable();            //根据hash值计算table的索引            else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {            //如果索引处为null,则直接在此处进行插入                if (casTabAt(tab, i, null, new Node
      
       (hash, key, value, null)))            //通过CAS进行插入，插入成功之后跳出循环                    break;                               }            else if ((fh = f.hash) == MOVED)            //当遇到表ForwardNode时，需要对表进行整合                tab = helpTransfer(tab, f);            else {                V oldVal = null;                //采用同步代码块进行上锁                synchronized (f) {                    if (tabAt(tab, i) == f) {                    //普通节点                        if (fh >= 0) {                            binCount = 1;                            //遍历所有节点                            for (Node
       
         e = f;; ++binCount) {                                K ek;                           //如果hash值跟key值都相同，则进行值替换            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) { //判断链表长度，如果链表长度大于阈值，进行树节点转换 if (binCount >= TREEIFY_THRESHOLD) treeifyBin(tab, i); if (oldVal != null) return oldVal; break; } } } addCount(1L, binCount);//改变table中的元素的个数 return null; }复制代码

下面分解一下步骤：

1.计算记录的key的hashCode，然后通过hashcode计算table的index位置

2.获取table[index]，为null直接采用casTabAt插入一条新的记录

3.不为null，则锁住根节点，然后通过节点的hashcode值来区分是链表还是红黑树

4.遍历链表或者红黑树，找到相应的节点，进行更新操作

5.如果是更新操作，则不需要进行transfer操作，否则判断一下更新之后的size有没有超过阈值，主要是判断链表的节点数跟table的数组长度

如何保证线程安全

当对table中的某个节点进行写操作的时候，如果为null,直接进行插入，注意如果不为null，则用syncronized关键字锁住当前节点，然后进行写操作。为什么为空的时候插入不需要上锁，因为ConcurrentHashMap采用的是CAS的乐观锁机制，如果有线程在进行修改，就会在外面等待，然后过一会儿再过来查询一下，这样可以保证高效并发。

get方法

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;    }复制代码

1.通过计算key的hash值进而计算key在table中的index

2.拿到table[index],为null,直接返回null

3.如果table[index]不为null，通过Node的hash值来判断是链表还是红黑树

4.遍历到相应的value之后直接返回，否则返回null

如何保证线程安全 首先当多个线程同时读取的时候，ConcurrentHashMap给每个Node节点加了volatile关键字修饰来保证可见性。

总结

看了很多数据结构，感觉ConcurrentHashMap是最复杂的，里面涉及到了很多的CAS操作，以及硬件层的线程安全操作，分析了好几天，才弄懂，下面总结一下c的JDK1.7跟1.8的区别。

JDK 1.7

JDK1.8

对比分析

1.7采用链表，1.8使用红黑树来替换之前的链表，提高了查找的效率。

1.7版本锁的粒度是基于Segment的，采用的是分段锁，粒度是每个数组1.8的实现降低锁的粒度到每个Node节点，

1.7采用的Segment继承自ReentrantLock，自带锁，1.8的数据结构采用数组+链表+红黑树，使用了大量的CAS操作，提升了高并发时的性能，丢弃了分段锁的概念。

1.7采用的是ReentrantLock进行同步，1.8 采用了synchronized进行同步，其实一开始没理解，因为ReentrantLock比synchronized更加灵活，后来思考了一下觉得是因为锁的粒度进一步降低，在粗粒度加锁中ReentrantLock可能通过Condition来控制各个低粒度的边界，更加的灵活，而在低粒度中，Condition的优势不大。

概述