ConcurrentHashMap源码解读

---

ConcurrentHashmap

ConcurrentHashmap整体跟hashmap差不多，主要的区别是线程安全，并且不支持key和value为null
ConcurrentHashmap在插入元素的时候会使用synchronized，保证了写的线程安全。在扩容的时候，将表划分成一个个独立的区域，然后每个执行transfer的线程各自负责一个区域，在保证了线程安全的情况下能提高扩容的效率。

1、族谱

public class ConcurrentHashMap<K,V> extends AbstractMap<K,V>
    implements ConcurrentMap<K,V>, Serializable {

2、基础知识

concurrentHashmap中的基础变量，大部分跟hashmap中都是相同的，例如DEFAULT_CAPACITY，MAXIMUM_CAPACITY等，下面主要针对于不同的地方进行介绍

常量

//并发级别，新版本已经都不用了，主要是兼容旧版
private static final int DEFAULT_CONCURRENCY_LEVEL = 16; 

//在并发transfer的时候，需要操作的最小的索引范围，也就是所谓的segment长度
private static final int MIN_TRANSFER_STRIDE = 16;   

//用来生成生成戳的一个变量，生成戳会被包含在sizeCtl中
private static int RESIZE_STAMP_BITS = 16;

//能够参与扩容的最大的线程数量，必须小于32-RESIZE_STAMP_BITS个比特位
private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1;

//在sizeCtl中帮忙记录大小的偏移位
private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;

//后面计算hash时候会用到的一些辅助常量
 static final int MOVED     = -1; // hash for forwarding nodes
 static final int TREEBIN   = -2; // hash for roots of trees
 static final int RESERVED  = -3; // hash for transient reservations
 static final int HASH_BITS = 0x7fffffff; // usable bits of normal node hash

//cpu数量
static final int NCPU = Runtime.getRuntime().availableProcessors();

变量

//记录表中的元素数量 
private transient volatile long baseCount;
//一个很重要的概念，主要用来控制初始化和扩容的（个人觉得跟ThreadPoolExecutor中的ctl很像，都是用一个int来存储状态跟数量）。如果为负值，说明当前表正在扩容或者初始化。为-1的时候表示正在初始化，-(1+扩容线程数量n
//)表示有n个线程正在帮忙扩容。如果表为空，使用初始化时候的值，默认为0，初始化之后，存储的是之后扩容得阈值
 private transient volatile int sizeCtl;

//下一个需要执行transfer任务的起点
 private transient volatile int transferIndex;

辅助对象

static final class ForwardingNode<K,V> extends Node<K,V> {                                  //在执行转移操作时，会被插入到索引头部的一个节点
        final Node<K,V>[] nextTable;                                                            //存储新的表
        ForwardingNode(Node<K,V>[] tab) {                                                       //存储当前节点的情况，表示已经转移了
            super(MOVED, null, null, null);
            this.nextTable = tab;
        }
   ***
   }

3、初始化

public ConcurrentHashMap() {
    } 

public ConcurrentHashMap(int initialCapacity) {
        if (initialCapacity < 0)                                                //根据提供的容量，获取到离指定容量最近的2的n次方
            throw new IllegalArgumentException();
        int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?
                   MAXIMUM_CAPACITY :
                   tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));
        this.sizeCtl = cap;
    }

 //传入一个map，在初始化的时候会将传入map的所有值复制到新的map中去
public ConcurrentHashMap(Map<? extends K, ? extends V> m) {
        this.sizeCtl = DEFAULT_CAPACITY;
        putAll(m);
    }


public ConcurrentHashMap(int initialCapacity, float loadFactor) {
        this(initialCapacity, loadFactor, 1);
    }

//自定义并发线程数
public ConcurrentHashMap(int initialCapacity,
                             float loadFactor, int concurrencyLevel) {
        if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0)
            throw new IllegalArgumentException();
        if (initialCapacity < concurrencyLevel)   // Use at least as many bins           //如果初始容量比并发线程还少，修改最小容量为并发线程数量
            initialCapacity = concurrencyLevel;   // as estimated threads
        long size = (long)(1.0 + (long)initialCapacity / loadFactor);
        int cap = (size >= (long)MAXIMUM_CAPACITY) ?
            MAXIMUM_CAPACITY : tableSizeFor((int)size);
        this.sizeCtl = cap;                                             //修改容量
    }

4、putAll

在上面能够看到，初始化的时候如果传入了一个map，那么就会新建一个map，然后根据传入的map去复制其中的值

public void putAll(Map<? extends K, ? extends V> m) {
        tryPresize(m.size());
        for (Map.Entry<? extends K, ? extends V> e : m.entrySet())
            putVal(e.getKey(), e.getValue(), false);
    }

其中的putval也是我们在后面执行put的时候会用的一个函数，之后会说到，这里我们先说一下tryPresize()

5、tryPresize

该函数执行的时候，有两种情况，一种是putall时候，还有一种是单个bucket中的元素达到了转换树的阈值，但是表中的总元素数量还没有达到阈值，会调用该方法进行扩容

private final void tryPresize(int size) {                                       
   
   //给定初始容量，里面tableSizeFor另外一篇文章有说到，主要是获取到最近的一个2的n次方，这里主要是设定初始容量
        int c = (size >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY :
            tableSizeFor(size + (size >>> 1) + 1);                               
        int sc;     
  //如果小于0，说明正在扩容,大于等于0的时候，表示可以进行扩容
        while ((sc = sizeCtl) >= 0) {                                              
            Node<K,V>[] tab = table; int n;
          	//如果当前table为0，也就是初始化
            if (tab == null || (n = tab.length) == 0) {     
              	//选择sizeCtl和size中比较大的一个
                n = (sc > c) ? sc : c;          
              //比较当前的sizeCtl是否一致，如果不一致，说明有线程进行了修改。在初始化的时候，该值会被置为-1，同时只会有一个线程进行该操作
                if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {                  
                    try {
                      //如果当前线程拿到了初始化的锁，并且table还没有被初始化，开始初始化数组
                        if (table == tab) {                                        
                            @SuppressWarnings("unchecked")
                            Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                            table = nt;
                          //初始化完成后，n >>> 2等于0.25n，n-(n>>>2) = 0.75n，也就是loadFactor，将sc修改为下一次扩容的阈值
                            sc = n - (n >>> 2);                                    
                        }
                    } finally {
                        sizeCtl = sc;
                    }
                }
            }
          //如果需要扩容的容量比sc还要小，说明不需要扩容。容量超过了最大容量，就跳出循环
            else if (c <= sc || n >= MAXIMUM_CAPACITY)                                  
                break;
            //扩容操作，本来要转换成红黑数的，key数量小于设定数量，就先执行扩容操作
            else if (tab == table) {     
              
                int rs = resizeStamp(n);
              //如果现在已经开始扩容了
                if (sc < 0) {                                                           
                    Node<K,V>[] nt;
               //这里的rs后面会说，rs+1,表示当前正在扩容，sc >>> RESIZE_STAMP_SHIFT) != rs表示当前的n发生了变化，sc == rs + MAX_RESIZERS表示当前参与扩容的线程已经达到了最大值，(nt = nextTable) == null表示扩容已经结束了，transferIndex <= 0表示没有transfer任务需要领取了
                    if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||            
                        sc == rs + MAX_RESIZERS || (nt = nextTable) == null ||          
                        transferIndex <= 0)
                        break;
                  //尝试获取一次transfer任务，如果获取成功，就对sc+1，
                    if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))                 //尝试去增加一个参数transfer的线程，先对sizeCtl加一，然后执行transfer
                        transfer(tab, nt);
                }
              //执行到这一步的条件是sc >=0,说明还没有开始transfer，尝试成为第一个transfer线程，
                else if (U.compareAndSwapInt(this, SIZECTL, sc,
                                             (rs << RESIZE_STAMP_SHIFT) + 2))           
                    transfer(tab, null);
            }
        }
    }

6、resizeStamp

/**
   * 返回长度为n时候的扩容所需要用到的生成戳
   * 在左移RESIZE_STAMP_SHIFT位之后，一定要是个负值
   */
  static final int resizeStamp(int n) {
      return Integer.numberOfLeadingZeros(n) | (1 << (RESIZE_STAMP_BITS - 1));            //numberOfLeadingZeros获得n前面的0的位数，将1左移15位，然后执行或运算
  }

7、transfer

transfer是在扩容完后，把旧的元素移动或者复制到新的表中

private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {                             
      int n = tab.length, stride;
  // 这里计算出需要的每个transfer所需要移动的bucket的数量
      if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
          stride = MIN_TRANSFER_STRIDE;   
  // nexttTab为空，说明是第一个开始执行transfer的线程
      if (nextTab == null) {                                                                      
          try {
              @SuppressWarnings("unchecked")
            //新建一个长度为之前2倍的数组，做为新数组
              Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];                                
              nextTab = nt;
          } catch (Throwable ex) {      // try to cope with OOME
            //如果发生oom，就把sizeCtl改为int的最大值。
              sizeCtl = Integer.MAX_VALUE;
              return;                                                                             
          }
          nextTable = nextTab;
        //在第一次transfer的时候，需要transfer的起点位置
          transferIndex = n;                                                                      
      }
  //新表的长度
      int nextn = nextTab.length;
  //先插入fwn，表示转移的状态
      ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);       
  // 表示是否需要处理下一个元素
      boolean advance = true;   
  // 表示当前是否已经处理完成
      boolean finishing = false; 
  //bound表示的当前transfer的segment的终点，每次transfer都是倒序执行的
      for (int i = 0, bound = 0;;) {
          Node<K,V> f; int fh;
        //判断是否有新的segment需要处理
          while (advance) {                                                                       
              int nextIndex, nextBound;
            //表示当前任务正在执行还未结束，--i >= bound，说明还没有处理到边界，如果已经处理到了，--i < bound，因为有任务还在处理，所以就不领取新的transfer任务
              if (--i >= bound || finishing)                                                      
                  advance = false;
            //没有任务可以领取了，可以退出transfer了
              else if ((nextIndex = transferIndex) <= 0) {                                        
                  i = -1;
                  advance = false;
              }
            //计算下一个需要处理的segment的起点,如果下一个需要处理的起点币stride小，说明剩下的元素已经小于stride了，需要把终点设置为0
              else if (U.compareAndSwapInt
                       (this, TRANSFERINDEX, nextIndex,                                           
                        nextBound = (nextIndex > stride ?                                         
                                     nextIndex - stride : 0))) {
                //设置要处理的segment的区间，bound表示终点，nextIndex为起点
                  bound = nextBound;
                  i = nextIndex - 1;                                                              
                  advance = false;
              }
          }                                                                                       //如果i坐标小于0，或者i超过了表的长度，或者i+旧表长度超过新表长度，说明已经处理结束了
          if (i < 0 || i >= n || i + n >= nextn) {
              int sc;
              if (finishing) {                                                                    //如果已经结束了，设置nextTable为空，避免被再次操作
                  nextTable = null;
                  table = nextTab;
                // n << 1 = 2n, n>>>1 = 0.5n sizeCtl = 1.5n，而现在已经扩容两倍了，相当于sizeCtl等于新表空间的0.75
                  sizeCtl = (n << 1) - (n >>> 1);                                                 
                  return;
              }
            //cas操作比较sizeCtl是否相等，如果相等，进行sizeCtl-1，表示将当前线程退出扩容，因为sizeCtl控制的是控制扩容的线程的数量
              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，表示已经转移过了，此时hash为MOVED
              advance = casTabAt(tab, i, null, fwd);
          else if ((fh = f.hash) == MOVED)                                                    //如果取到当前节点，并且上面的hash值为MOVED，说明已经处理过了
              advance = true; // already processed
          else {
              synchronized (f) {                                                              //到这一步，就要开始实际处理了
                  if (tabAt(tab, i) == f) {                                                   //二次校验，判断当前元素是否发生了变化
                      Node<K,V> ln, hn;
                      if (fh >= 0) {                                                          //如果没有forwardingNode，正常情况下hash都是大于0的
                          int runBit = fh & n;                                                //这个东西类似于hashMap里面的e.hash & oldCap，也就是计算需不需要移动到新的索引位置上去
                          Node<K,V> lastRun = f;                                              //下面会复用其中的某一个链表，然后找到该链表的最后一个值
                          for (Node<K,V> p = f.next; p != null; p = p.next) {
                              int b = p.hash & n;
                              if (b != runBit) {
                                  runBit = b;
                                  lastRun = p;
                              }
                          }
                          if (runBit == 0) {                                                  //如果为0，说明不需要移位
                              ln = lastRun;
                              hn = null;
                          }
                          else {
                              hn = lastRun;
                              ln = null;
                          }
                          for (Node<K,V> 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<K,V>(ph, pk, pv, ln);
                              else
                                  hn = new Node<K,V>(ph, pk, pv, hn);
                          }
                          setTabAt(nextTab, i, ln);                                           //把bucket处理完之后，把需要移位以及不需要移位的元素，都加到新表去
                          setTabAt(nextTab, i + n, hn);
                          setTabAt(tab, i, fwd);                                              //给旧表的对应位置加一个forwardingNode
                          advance = true;                                                     //当前处理完了，需要处理下一个
                      }
                      else if (f instanceof TreeBin) {                                        //如果当前bucket是红黑树
                          TreeBin<K,V> t = (TreeBin<K,V>)f;
                          TreeNode<K,V> lo = null, loTail = null;
                          TreeNode<K,V> hi = null, hiTail = null;
                          int lc = 0, hc = 0;
                          for (Node<K,V> e = t.first; e != null; e = e.next) {                //计算
                              int h = e.hash;
                              TreeNode<K,V> p = new TreeNode<K,V>
                                  (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;
                              }
                          }
                          ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) :              //如果树的深度小于设定的阈值，则将树转换为链表，如果
                              (hc != 0) ? new TreeBin<K,V>(lo) : t;                       //如果所有元素都不需要移位，直接使用上面初始化的对象，否则新建树去装载
                          hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) :
                              (lc != 0) ? new TreeBin<K,V>(hi) : t;
                          setTabAt(nextTab, i, ln);
                          setTabAt(nextTab, i + n, hn);
                          setTabAt(tab, i, fwd);
                          advance = true;
                      }
                  }
              }
          }
      }
  }

8、helpTransfer

该函数的用处较多，用途是为了帮助进行扩容过程中的transfer，在putVal的时候，也会执行该函数

final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
        Node<K,V>[] nextTab; int sc;
        if (tab != null && (f instanceof ForwardingNode) &&             //如果表不为空，并且f是一个forwardingNode（说明正在移位），并且新的表并不为空，则开始帮助transfer
            (nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) {
          //根据table长度获得一个生成戳
            int rs = resizeStamp(tab.length);                           
          //如果在执行的时候，nextTable和table还没有被别的更改并且sc<0，也就是处于扩容状态
            while (nextTab == nextTable && table == tab &&              
                   (sc = sizeCtl) < 0) {
              // sc == rs + 1表示当前正在扩容  sc >>> RESIZE_STAMP_SHIFT) != rs表示当前的sc计算之后得到的rs不一样，也就是说n发生了变化  transferIndex <=0 说明没有需要帮助转移的任务了 sc == rs + MAX_RESIZERS表示当前已经达到最大的转移线程了
                if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 || 
                    sc == rs + MAX_RESIZERS || transferIndex <= 0)     
                    break;                                               
                if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
                  //在sc里面加一，相当于领取一个任务，然后开始transfer
                    transfer(tab, nextTab);                             
                    break;
                }
            }
            return nextTab;
        }
        return table;
    }

9、initTable

在第一次put的时候，如果表还没有初始化，会调用

private final Node<K,V>[] initTable() {
       Node<K,V>[] tab; int sc;
       while ((tab = table) == null || tab.length == 0) {
           if ((sc = sizeCtl) < 0)                                     //初始化map，如果没有拿到锁，就先自旋等待
               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<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
                       table = tab = nt;
                       sc = n - (n >>> 2);                             //设定阈值为0.75n
                   }
               } finally {
                   sizeCtl = sc;
               }
               break;
           }
       }
       return tab;
   }

10、putVal

实际的插入函数

final V putVal(K key, V value, boolean onlyIfAbsent) {
  //校验key或者value是否为空,跟hashmap的区别的，hashmap的key可以为空，concurrentHashmap的key和value都不能为空
       if (key == null || value == null) throw new NullPointerException();
  //计算key的hashCode，跟hashmap里面的计算规则一样
       int hash = spread(key.hashCode());                               
       int binCount = 0;
  //对表循环操作，找到要操作的点
       for (Node<K,V>[] tab = table;;) {                               
           Node<K,V> f; int n, i, fh;
         //如果表还没有初始化，调用initTable对表进行初始化
           if (tab == null || (n = tab.length) == 0)                   
               tab = initTable();
         //如果要插入的索引位置还没有元素
           else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {    
             // 使用cas操作，将新节点插入到索引位置去，该操作没有锁
               if (casTabAt(tab, i, null,
                            new Node<K,V>(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;  
             //使用同步操作，来获取最新的f
               synchronized (f) {        
                 //如果当前索引上面的值没有发生变化，因为在这期间可能有线程对f进行了操作，就会导致索引上面的对象发生了变化
                   if (tabAt(tab, i) == f) {             
                     //fh 为f.hash,如果f.hash>0,说明至少有一个元素在上面
                       if (fh >= 0) {                                  
                           binCount = 1;
                           for (Node<K,V> e = f;; ++binCount) {
                               K ek;
                             //如果key相同，hash相同，那么就替换原有的value值，然后把旧值返回去
                               if (e.hash == hash &&
                                   ((ek = e.key) == key ||             
                                    (ek != null && key.equals(ek)))) {
                                   oldVal = e.val;
                                   if (!onlyIfAbsent)
                                       e.val = value;
                                   break;
                               }
                               Node<K,V> pred = e;
                             //如果e.next == null说明是bucket的最后一个节点了，就把新的节点加入到最后一个节点后面
                               if ((e = e.next) == null) {             
                                   pred.next = new Node<K,V>(hash, key,
                                                             value, null);
                                   break;
                               }
                           }
                       }
                     //如果节点是红黑树
                       else if (f instanceof TreeBin) {                
                           Node<K,V> p;
                         //因为已经是树了，而binCount主要是拿来做转换成树的操作的，所以随便给个值就好了
                           binCount = 2;                 
                          //向树中添加节点，如果原来有相同key的节点存在，替换旧值，并返回旧值
                           if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
                                                          value)) != null) {
                               oldVal = p.val;                        
                               if (!onlyIfAbsent)
                                   p.val = value;
                           }
                       }
                   }
               }
              //如果bucket的容量大于0
               if (binCount != 0) {          
                 //如果达到了转换成树的阈值。执行转换操作
                   if (binCount >= TREEIFY_THRESHOLD)          
                       treeifyBin(tab, i);
                 //如果put之前的值不为空，返回旧值
                   if (oldVal != null)                         
                       return oldVal;
                   break;
               }
           }
       }
  			//计数器增加
       addCount(1L, binCount);
       return null;
   }