并发编程-容器ConcurrentHashMap源码分析

1.重要的属性

首先来看几个重要的属性，与HashMap相同的就不再介绍了，这里重点解释一下sizeCtl这个属性。可以说它是ConcurrentHashMap中出镜率很高的一个属性，因为它是一个控制标识符，在不同的地方有不同用途，而且它的取值不同，也代表不同的含义。

• 负数代表正在进行初始化或扩容操作
• -1代表正在初始化
• -N 表示有N-1个线程正在进行扩容操作
• 0代表hash表还没有被初始化，
• 正数表示这个初始化或下一次进行扩容的大小，这一点类似于扩容阈值的概念。还后面可以看到，它的值始终是当前ConcurrentHashMap容量的0.75倍，这与loadfactor是对应的。

  //盛装Node元素的数组,它的大小是2的整数次幂
  transient volatile Node<K,V>[] table;
  
  /**
   hash表初始化或扩容时的一个控制位标识量。
   负数代表正在进行初始化或扩容操作
   -1代表正在初始化
   -N 表示有N-1个线程正在进行扩容操作
   正数或0代表hash表还没有被初始化，这个数值表示初始化或下一次进行扩容的大小
   */
  private transient volatile int sizeCtl;
  
  // 以下两个是用来控制扩容的时候 单线程进入的变量
   /**
   * The number of bits used for generation stamp in sizeCtl.
   * Must be at least 6 for 32bit arrays.
   */
  private static int RESIZE_STAMP_BITS = 16;
  /**
   * The bit shift for recording size stamp in sizeCtl.
   */
  private static final int RESIZE_STAMP_SHIFT = 32- RESIZE_STAMP_BITS;
  
  static final int MOVED     = -1;// hash值是-1，表示这是一个forwardNode节点
  static final int TREEBIN   = -2;// hash值是-2  表示这时一个TreeBin节点

2.重要的类

2.1 Node

Node是最核心的内部类，它包装了key-value键值对，所有插入ConcurrentHashMap的数据都包装在这里面。它与HashMap中的定义很相似，但是但是有一些差别它对value和next属性设置了volatile同步锁(与JDK7的Segment相同)，它不允许调用setValue方法直接改变Node的value域，它增加了find方法辅助map.get()方法。

2.2 TreeNode

树节点类，另外一个核心的数据结构。当链表长度过长的时候，会转换为TreeNode。但是与HashMap不相同的是，它并不是直接转换为红黑树，而是把这些结点包装成TreeNode放在TreeBin对象中，由TreeBin完成对红黑树的包装。而且TreeNode在ConcurrentHashMap继承自Node类，而不像HashMap中的继承自LinkedHashMap.Entry<K,V>类，也就是说TreeNode带有next指针，这样做的目的是方便基于TreeBin的访问。

2.3 TreeBin

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

这里仅贴出它的构造方法。可以看到在构造TreeBin节点时，仅仅指定了它的hash值为TREEBIN常量，这也就是个标识为。同时也看到我们熟悉的红黑树构造方法

2.4 ForwardingNode

一个用于连接两个table的节点类。它包含一个nextTable指针，用于指向下一张表。而且这个节点的key value next指针全部为null，它的hash值为-1。这里面定义的find的方法是从nextTable里进行查询节点，而不是以自身为头节点进行查找。

/**
 * A node inserted at head of bins during transfer operations.
 */
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;
    }

    Node<K,V> find(inth, Object k) {
        // loop to avoid arbitrarily deep recursion on forwarding nodes
        outer:for(Node<K,V>[] tab = nextTable;;) {
            Node<K,V> e; intn;
            if(k == null|| tab == null|| (n = tab.length) == 0||
                (e = tabAt(tab, (n - 1) & h)) == null)
                returnnull;
            for(;;) {
                inteh; K ek;
                if((eh = e.hash) == h &&
                    ((ek = e.key) == k || (ek != null&& k.equals(ek))))
                    returne;
                if(eh < 0) {
                    if(einstanceofForwardingNode) {
                        tab = ((ForwardingNode<K,V>)e).nextTable;
                        continueouter;
                    }
                    else
                        returne.find(h, k);
                }
                if((e = e.next) == null)
                    returnnull;
            }
        }
    }
}

3.Unsafe与CAS

在ConcurrentHashMap中，随处可以看到U, 大量使用了U.compareAndSwapXXX的方法，这个方法是利用一个CAS算法实现无锁化的修改值的操作，他可以大大降低锁代理的性能消耗。这个算法的基本思想就是不断地去比较当前内存中的变量值与你指定的一个变量值是否相等，如果相等，则接受你指定的修改的值，否则拒绝你的操作。因为当前线程中的值已经不是最新的值，你的修改很可能会覆盖掉其他线程修改的结果。这一点与乐观锁，SVN的思想是比较类似的。

3.1 unsafe静态块

unsafe代码块控制了一些属性的修改工作，比如最常用的SIZECTL 。在这一版本的concurrentHashMap中，大量应用来的CAS方法进行变量、属性的修改工作。利用CAS进行无锁操作，可以大大提高性能。

private static final sun.misc.Unsafe U;
   private static final long SIZECTL;
   private static final long TRANSFERINDEX;
   private static final long BASECOUNT;
   private static final long CELLSBUSY;
   private static final long CELLVALUE;
   private static final long ABASE;
   private static final int ASHIFT;
 
   static{
       try{
           U = sun.misc.Unsafe.getUnsafe();
           Class<?> k = ConcurrentHashMap.class;
           SIZECTL = U.objectFieldOffset(k.getDeclaredField("sizeCtl"));
           TRANSFERINDEX = U.objectFieldOffset(k.getDeclaredField("transferIndex"));
           BASECOUNT = U.objectFieldOffset(k.getDeclaredField("baseCount"));
           CELLSBUSY = U.objectFieldOffset(k.getDeclaredField("cellsBusy"));
           Class<?> ck = CounterCell.class;
           CELLVALUE = U.objectFieldOffset(ck.getDeclaredField("value"));
           Class<?> ak = Node[].class;
           ABASE = U.arrayBaseOffset(ak);
           intscale = U.arrayIndexScale(ak);
           if((scale & (scale - 1)) != 0)
               thrownewError("data type scale not a power of two");
           ASHIFT = 31- Integer.numberOfLeadingZeros(scale);
       }catch(Exception e) {
           thrownewError(e);
       }
   }

3.2 三个核心方法

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

static final <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i) {
		//获得在i位置上的Node节点,
    return(Node<K,V>)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE);
}  
static final <K,V> boolean casTabAt(Node<K,V>[] tab, int i, Node<K,V> c, Node<K,V> v) {
    //因此当前线程中的值并不是最新的值，这种修改可能会覆盖掉其他线程的修改结果有点类似于SVN
    returnU.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v);
}
static final <K,V> voidsetTabAt(Node<K,V>[] tab, int i, Node<K,V> v) {
    //利用volatile方法设置节点位置的值
    U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v);
}

tabAt操作能够实时感知其他线程对map的修改，它调用了native方法getObjectVolatile，该方法获取到的总是最新的值。

casTabAt通过CAS修改table中的值

4 初始化方法initTable

对于ConcurrentHashMap来说，调用它的构造方法仅仅是设置了一些参数而已。而整个table的初始化是在向ConcurrentHashMap中插入元素的时候发生的。如调用put、computeIfAbsent、compute、merge等方法的时候，调用时机是检查table==null。

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

private final Node<K,V>[] initTable() {
    Node<K,V>[] tab; 
    int sc;
    while((tab = table) == null|| tab.length == 0) {
        //sizeCtl表示有其他线程正在进行初始化操作，把线程挂起。对于table的初始化工作，只能有一个线程在进行。
        if((sc = sizeCtl) < 0)
            Thread.yield();
        else if(U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
            //利用CAS方法把sizectl的值置为-1 表示本线程正在进行初始化
            try{
                if((tab = table) == null|| tab.length == 0) {
                    intn = (sc > 0) ? sc : DEFAULT_CAPACITY;
                    @SuppressWarnings("unchecked")
                    Node<K,V>[] nt = (Node<K,V>[])newNode<?,?>[n];
                    table = tab = nt;
                    sc = n - (n >>> 2);//相当于0.75*n 设置一个扩容的阈值
                }
            }finally{
                sizeCtl = sc;
            }
            break;
        }
    }
    returntab;
}

5 扩容方法 transfer

当ConcurrentHashMap容量不足的时候，需要对table进行扩容。这个方法的基本思想跟HashMap是很像的，但是由于它是支持并发扩容的，所以要复杂的多。原因是它支持多线程进行扩容操作，而并没有加锁。我想这样做的目的不仅仅是为了满足concurrent的要求，而是希望利用并发处理去减少扩容带来的时间影响。因为在扩容的时候，总是会涉及到从一个“数组”到另一个“数组”拷贝的操作，如果这个操作能够并发进行，那真是太好了。

整个扩容操作分为两个部分

• 第一部分是构建一个nextTable，它的容量是原来的两倍，这个操作是单线程完成的。这个单线程的保证是通过RESIZE_STAMP_SHIFT这个常量经过一次运算来保证的，这个地方在后面会有提到；
• 第二个部分就是将原来table中的元素复制到nextTable中，这里允许多线程进行操作。

先来看一下单线程是如何完成的：

它的大体思想就是遍历、复制的过程。首先根据运算得到需要遍历的次数n，然后利用tabAt方法获得i位置的元素：

1. 如果这个位置为空，就在原table中的i位置放入ForwardNode节点，这个也是触发并发扩容的关键点；
2. 如果这个位置是Node节点（fh>=0），且它是一个链表的头节点，就构造一个反序链表，把他们分别放在nextTable的i和i+n的位置上
3. 如果这个位置是TreeBin节点（fh<0），也做一个反序处理，并且判断是否需要untreefi，把处理的结果分别放在nextTable的i和i+n的位置上
4. 遍历过所有的节点以后就完成了复制工作，这时让nextTable作为新的table，并且更新sizeCtl为新容量的0.75倍 ，完成扩容。
   再看一下多线程是如何完成的：

在代码的69行有一个判断，如果遍历到的节点是forward节点，就向后继续遍历，再加上给节点上锁的机制，就完成了多线程的控制。多线程遍历节点，处理了一个节点，就把对应点的值set为forward，另一个线程看到forward，就向后遍历。这样交叉就完成了复制工作。而且还很好的解决了线程安全的问题。 这个方法的设计实在是让人膜拜。

6 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. 如果检测到要插入的节点是非空且不是forward节点，就对这个节点加锁，这样就保证了线程安全。尽管这个有一些影响效率，但是还是会比对整个hashTable的synchronized要好得多。

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

如果这个位置是空的，那么直接放入，而且不需要加锁操作。

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

publicV put(K key, V value) {
    return putVal(key, value, false);
}

/** Implementation for put and putIfAbsent */
final V putVal(K key, V value, boolean onlyIfAbsent) {
    if(key == null|| value == null)
        throw new NullPointerException();
    //计算hash值
    int hash = spread(key.hashCode());
    int binCount = 0;
    //死循环 何时插入成功 何时跳出
    for(Node<K,V>[] tab = table;;) {
        Node<K,V> f; 
        int n, i, fh;
        if(tab == null|| (n = tab.length) == 0)
          	//如果table为空的话，初始化table >>
            tab = initTable();
        else if((f = tabAt(tab, i = (n - 1) & hash)) == null) {
            //根据hash值计算出在table里面的位置，如果这个位置没有值，直接放进去，不需要加锁
            if(casTabAt(tab, i, null,new Node<K,V>(hash, key, value, null)))
                break;  
        }
        ...
}

假如在上面这段代码中存在两个线程，在不加锁的情况下：线程 A 成功执行 casTabAt 操作后，随后的线程 B 可以通过 tabAt 方法立刻看到 table[i]的改变。原因如下：线程 A 的casTabAt 操作，具有 volatile 读写相同的内存语义，根据 volatile 的 happens-before 规则：线程 A 的 casTabAt 操作，一定对线程 B 的 tabAt 操作可见。

我们可以发现JDK8中的实现也是锁分离的思想，只是锁住的是一个Node，而不是JDK7中的Segment，而锁住Node之前的操作是无锁的并且也是线程安全的，建立在之前提到的3个原子操作上。

6.1 helpTransfer方法

这是一个协助扩容的方法。这个方法被调用的时候，当前ConcurrentHashMap一定已经有了nextTable对象，首先拿到这个nextTable对象，调用transfer方法。回看上面的transfer方法可以看到，当本线程进入扩容方法的时候会直接进入复制阶段。

6.2 treeifyBin方法

这个方法用于将过长的链表转换为TreeBin对象。但是他并不是直接转换，而是进行一次容量判断，如果容量没有达到转换的要求，直接进行扩容操作并返回；如果满足条件才链表的结构抓换为TreeBin ，这与HashMap不同的是，它并没有把TreeNode直接放入红黑树，而是利用了TreeBin这个小容器来封装所有的TreeNode.

7 get方法

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

public V get(Object key) {
    Node<K,V>[] tab; 
    Node<K,V> e, p; 
    int n, eh; 
    K ek;
    //计算hash值
    int h = spread(key.hashCode());
    //根据hash值确定节点位置
    if((tab = table) != null && (n = tab.length) > 0 && (e = tabAt(tab, (n - 1) & h)) != null) {
        //如果搜索到的节点key与传入的key相同且不为null,直接返回这个节点 
        if((eh = e.hash) == h) {
            if((ek = e.key) == key || (ek != null&& key.equals(ek)))
                returne.val;
        }
        //如果eh<0 说明这个节点在树上 直接寻找
        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;
}

8 Size相关的方法

对于ConcurrentHashMap来说，这个table里到底装了多少东西其实是个不确定的数量，因为不可能在调用size()方法的时候像GC的“stop the world”一样让其他线程都停下来让你去统计，因此只能说这个数量是个估计值。对于这个估计值，ConcurrentHashMap也是大费周章才计算出来的。

8.1 辅助定义

为了统计元素个数，ConcurrentHashMap定义了一些变量和一个内部类

/**
 * A padded cell for distributing counts.  Adapted from LongAdder
 * and Striped64.  See their internal docs for explanation.
 */
@sun.misc.Contendedstaticfinalclass CounterCell {
    volatilelongvalue;
    CounterCell(longx) { value = x; }
}

/******************************************/ 

/**
 * 实际上保存的是hashmap中的元素个数  利用CAS锁进行更新
 但它并不用返回当前hashmap的元素个数

 */
privatetransientvolatile long baseCount;
/**
 * Spinlock (locked via CAS) used when resizing and/or creating CounterCells.
 */
privatetransientvolatile int cellsBusy;

/**
 * Table of counter cells. When non-null, size is a power of 2.
 */
privatetransientvolatile CounterCell[] counterCells;

8.2 mappingCount与Size方法

mappingCount与size方法的类似 从Java工程师给出的注释来看，应该使用mappingCount代替size方法 两个方法都没有直接返回basecount 而是统计一次这个值，而这个值其实也是一个大概的数值，因此可能在统计的时候有其他线程正在执行插入或删除操作。

publicintsize() {
        longn = sumCount();
        return((n < 0L) ? 0:
                (n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :
                (int)n);
    }
     /**
     * Returns the number of mappings. This method should be used
     * instead of {@link #size} because a ConcurrentHashMap may
     * contain more mappings than can be represented as an int. The
     * value returned is an estimate; the actual count may differ if
     * there are concurrent insertions or removals.
     *
     * @return the number of mappings
     * @since 1.8
     */
    publiclongmappingCount() {
        longn = sumCount();
        return(n < 0L) ? 0L : n; // ignore transient negative values
    }
 
     finallongsumCount() {
        CounterCell[] as = counterCells; CounterCell a;
        longsum = baseCount;
        if(as != null) {
            for(inti = 0; i < as.length; ++i) {
                if((a = as[i]) != null)
                    sum += a.value;//所有counter的值求和
            }
        }
        returnsum;
    }

8.3 addCount方法

在put方法结尾处调用了addCount方法，把当前ConcurrentHashMap的元素个数+1这个方法一共做了两件事,更新baseCount的值，检测是否进行扩容。

privatefinalvoid addCount(longx,intcheck) {
    CounterCell[] as; longb, s;
    //利用CAS方法更新baseCount的值
    if((as = counterCells) != null||
        !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {
        CounterCell a; longv;intm;
        booleanuncontended = true;
        if(as == null|| (m = as.length - 1) < 0||
            (a = as[ThreadLocalRandom.getProbe() & m]) == null||
            !(uncontended =
              U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) {
            fullAddCount(x, uncontended);
            return;
        }
        if(check <= 1)
            return;
        s = sumCount();
    }
    //如果check值大于等于0 则需要检验是否需要进行扩容操作
    if(check >= 0) {
        Node<K,V>[] tab, nt; intn, sc;
        while(s >= (long)(sc = sizeCtl) && (tab = table) != null&&
               (n = tab.length) < MAXIMUM_CAPACITY) {
            intrs = resizeStamp(n);
            //
            if(sc < 0) {
                if((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1||
                    sc == rs + MAX_RESIZERS || (nt = nextTable) == null||
                    transferIndex <= 0)
                    break;
                 //如果已经有其他线程在执行扩容操作
                if(U.compareAndSwapInt(this, SIZECTL, sc, sc + 1))
                    transfer(tab, nt);
            }
            //当前线程是唯一的或是第一个发起扩容的线程  此时nextTable=null
            elseif(U.compareAndSwapInt(this, SIZECTL, sc,
                                         (rs << RESIZE_STAMP_SHIFT) + 2))
                transfer(tab,null);
            s = sumCount();
        }
    }
}

总结

JDK6,7中的ConcurrentHashmap主要使用Segment来实现减小锁粒度，把HashMap分割成若干个Segment，在put的时候需要锁住Segment，get时候不加锁，使用volatile来保证可见性，当要统计全局时（比如size），首先会尝试多次计算modcount来确定，这几次尝试中，是否有其他线程进行了修改操作，如果没有，则直接返回size。如果有，则需要依次锁住所有的Segment来计算。

jdk7中ConcurrentHashmap中，当长度过长碰撞会很频繁，链表的增改删查操作都会消耗很长的时间，影响性能,所以jdk8 中完全重写了concurrentHashmap,代码量从原来的1000多行变成了 6000多 行，实现上也和原来的分段式存储有很大的区别。

主要设计上的变化有以下几点:

1. 不采用segment而采用node，锁住node来实现减小锁粒度。
2. 设计了MOVED状态 当resize的中过程中 线程2还在put数据，线程2会帮助resize。
3. 使用3个CAS操作来确保node的一些操作的原子性，这种方式代替了锁。
4. sizeCtl的不同值来代表不同含义，起到了控制的作用。

至于为什么JDK8中使用synchronized而不是ReentrantLock，我猜是因为JDK8中对synchronized有了足够的优化吧。

---

参考文档

JDK1.8 实现解读
扩容源码分析