JDK8中concorrentHashMap源码解析

1、构造方法

ConcurrentHashMap()

   /**
    * Creates a new, empty map with the default initial table size (16).
    */
/* 使用默认的初始表大小（16）创建一个新的空Map。 */
   public ConcurrentHashMap() {
   }

ConcurrentHashMap(int initialCapacity)

   /**
    * Creates a new, empty map with an initial table size
    * accommodating the specified number of elements without the need
    * to dynamically resize.
    *
    * @param initialCapacity The implementation performs internal
    * sizing to accommodate this many elements.
    * @throws IllegalArgumentException if the initial capacity of
    * elements is negative
    */
/**
    * 创建一个具有初始表大小的新的空映射*容纳指定数量的元素，而无需
    * 动态调整大小。 
    * 
    * @param initialCapacity该实现执行内部大小调整以容纳许多元素。 
    * 如果元素的初始容量为负，则抛出IllegalArgumentException 
    */
   public ConcurrentHashMap(int initialCapacity) {
       if (initialCapacity < 0)
           throw new IllegalArgumentException();
       int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ?
                  MAXIMUM_CAPACITY :
                  tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1));
       this.sizeCtl = cap;
   }

ConcurrentHashMap(int initialCapacity, float loadFactor)

/**
  * Creates a new, empty map with an initial table size based on
  * the given number of elements ({@code initialCapacity}) and
  * initial table density ({@code loadFactor}).
  *
  * @param initialCapacity the initial capacity. The implementation
  * performs internal sizing to accommodate this many elements,
  * given the specified load factor.
  * @param loadFactor the load factor (table density) for
  * establishing the initial table size
  * @throws IllegalArgumentException if the initial capacity of
  * elements is negative or the load factor is nonpositive
  *
  * @since 1.6
  */
 public ConcurrentHashMap(int initialCapacity, float loadFactor) {
     this(initialCapacity, loadFactor, 1);
 }

ConcurrentHashMap(int initialCapacity, float loadFactor, int concurrencyLevel)

/**
 * Creates a new, empty map with an initial table size based on
 * the given number of elements ({@code initialCapacity}), table
 * density ({@code loadFactor}), and number of concurrently
 * updating threads ({@code concurrencyLevel}).
 *
 * @param initialCapacity the initial capacity. The implementation
 * performs internal sizing to accommodate this many elements,
 * given the specified load factor.
 * @param loadFactor the load factor (table density) for
 * establishing the initial table size
 * @param concurrencyLevel the estimated number of concurrently
 * updating threads. The implementation may use this value as
 * a sizing hint.
 * @throws IllegalArgumentException if the initial capacity is
 * negative or the load factor or concurrencyLevel are
 * nonpositive
 */
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;
}

ConcurrentHashMap(Map<? extends K, ? extends V> m)

/**
 * Creates a new map with the same mappings as the given map.
 *
 * @param m the map
 */
public ConcurrentHashMap(Map<? extends K, ? extends V> m) {
    this.sizeCtl = DEFAULT_CAPACITY;
    putAll(m);
}

2、put方法

put(K key, V value)

/**
 * Maps the specified key to the specified value in this table.
 * Neither the key nor the value can be null.
 *
 * <p>The value can be retrieved by calling the {@code get} method
 * with a key that is equal to the original key.
 *
 * @param key key with which the specified value is to be associated
 * @param value value to be associated with the specified key
 * @return the previous value associated with {@code key}, or
 *         {@code null} if there was no mapping for {@code key}
 * @throws NullPointerException if the specified key or value is null
 */
public V put(K key, V value) {
    return putVal(key, value, false);
}

putVal(K key, V value, boolean onlyIfAbsent)

/** Implementation for put and putIfAbsent */
final V putVal(K key, V value, boolean onlyIfAbsent) {
    if (key == null || value == null) throw new NullPointerException();
    int hash = spread(key.hashCode());
    int binCount = 0;
    for (Node<K,V>[] tab = table;;) {
        Node<K,V> f; int n, i, fh;
        // 如果table为空， 初始化table,
        if (tab == null || (n = tab.length) == 0)
            tab = initTable();
        else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
            // 判断当前hash 的位置有没有值，没有值， 直接使使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;
            // 锁住单个 node
            synchronized (f) {
                // 再次检查是否有变更
                if (tabAt(tab, i) == f) {
                    // 如果这个节点hash 值不为0， 意思是当前节点为普通节点的时候， 这里应该比较容易理解， 比较hash 值， key equals 是否相等， 如果hash 冲突就添加链表， 记录链表长度(binCount)，之后会根据长度调整， 是否使用红黑树代替链表
                    if (fh >= 0) {
                        binCount = 1;
                        for (Node<K,V> e = f;; ++binCount) {
                            K ek;
                            if (e.hash == hash &&
                                ((ek = e.key) == key ||
                                 (ek !=   && key.equals(ek)))) {
                                oldVal = e.val;
                                if (!onlyIfAbsent)
                                    e.val = value;
                                break;
                            }
                            Node<K,V> pred = e;
                            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 = 2;
                        if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
                                                       value)) != null) {
                            oldVal = p.val;
                            if (!onlyIfAbsent)
                                p.val = value;
                        }
                    }
                }
            }
            if (binCount != 0) {
                // 阀值，默认是8， 超过会转换成树
                if (binCount >= TREEIFY_THRESHOLD)
                    // 转化为 tree
                    treeifyBin(tab, i);
                if (oldVal != null)
                    return oldVal;
                break;
            }
        }
    }
    addCount(1L, binCount);
    return null;
}

小结：

1. 若数组空，则初始化，完成之后，转2
2. 计算当前桶位是否有值

• 无,则 CAS 创建,失败后继续自旋,直到成功
• 有,转3

3. 判断桶位是否为转移节点(扩容ing)

• 是,则一直自旋等待扩容完成,之后再新增
• 否,转4

4. 桶位有值,对当前桶位加synchronize锁

• 链表，新增节点到链尾
• 红黑树，红黑树版方法新增

5. 新增完成之后,检验是否需要扩容

initTable()

   /**
    * Initializes table, using the size recorded in sizeCtl.
    */
/* 使用sizeCtl中记录的大小初始化表 */
   private final Node<K,V>[] initTable() {
       Node<K,V>[] tab; int sc;
       while ((tab = table) == null || tab.length == 0) {
           // sizeCtl: table 初始化和resize的标志位，表初始化和调整大小控件。
           if ((sc = sizeCtl) < 0)
               // sizeCtl为负值时，将初始化或调整表的大小
               Thread.yield(); // lost initialization race; just spin(丢失了初始化竞赛；旋转一下)
           // 设置SIZECTL为-1，设置成功开始初始化， 不成功继续循环。
           // compareAndSwapInt 非阻塞同步原语： arg0, arg1, arg2, arg3 分别为对象实例，目标对象属性，当前预期值，要设的值, 设置成功返回 true, 失败 false
           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);
                   }
               } finally {
                   sizeCtl = sc;
               }
               break;
           }
       }
       return tab;
   }

小结：

执行第一次put操作的线程会执行Unsafe.compareAndSwapInt方法修改sizeCtl为-1，有且只有一个线程能够修改成功，而其它线程只能通过Thread.yield()让出CPU时间片等待table初始化完成。

3、图解流程

总结：

jdk7 和 jdk8 的差异

• jdk7 使用 ReentrantLock + segment + hashentry + unsafe
• jdk8 使用 Synchronized + CAS + Node + NodeTree + Unsafe

jdk8 总结

• jdk8 用 Synchronized + CAS + Node + NodeTree 代替 Segment ，只有在hash 冲突， 或者修改已经值的时候才去加锁， 锁的粒度更小，大幅减少阻塞
• jdk8 链表节点数量大于8时，会将链表转化为红黑树进行存储，查询时间复杂度从O(n)，变成遍历红黑树O(logN)。