jdk8中的ConcurrentHashMap原理和源码解读

HashMap本身是线程不安全的，所以jdk提供了ConcurrentHashMap,这一篇来看下jdk8中的实现

一、基本原理

在jdk7中采用了Segment分段锁的思想来实现，在jdk8中不再采用分段锁的思想，jdk8中的ConcurrentHashMap

和HashMap一样，都只有一层Entry数组来实现，那么它是怎么保证线程安全呢,我们通过一张图来说明

当两个线程需要操作同一个下标时，先使用cas操作保证只会给下标处创建一个entry对象,再用Synchronized给entry对象加锁进行put操作 ,也就是使用cas+synchronize来保证线程安全。

二、构造方法

因为不再使用分段锁，所以构造方法就简单多了,

//无参数的构造方法什么都不做，全部使用默认值
public ConcurrentHashMap() {
}

//有参数的构造方法只是计算了下map的初始容量，依旧是得到一个2的幂次方数
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;
}

三、put方法

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

//真正的put逻辑,这个方法看着很长，但实际就是先用cas操作得当一个Node对象,然后锁住这个对象进行put操作
 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;
            if (tab == null || (n = tab.length) == 0)
                //初始化map中的数组，这个操作内部也是用的cas保证只有一个线程能初始化成功
                tab = initTable();
            //取某个下标处的node,如果为null就需要用cas操作生成一个新的
            else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
                //这个if就是在用cas操作生成一个新的节点放到数组此下标处，如果成功了就退出循环,
                //如果没成功说明别的线程给这个下标处放了一个元素，就需要在下轮循环中使用
                //锁主此元素的方式进行put
                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)
                //扩容过程中会把当前处理过的数组下标处放一个ForwardingNode节点,
                //hash值是-1，别的线程put时根据这个标记发现当前在扩容就会帮助扩容
                tab = helpTransfer(tab, f);
            else {
                V oldVal = null;
                //走到这里表示某个下标处已经有元素了，然后给这元素加锁,继续把要put的元素加入到
                //这个链表或者红黑树上。
                synchronized (f) {
                    if (tabAt(tab, i) == f) {
                        if (fh >= 0) {
                            binCount = 1;
                            for (Node<K,V> e = f;; ++binCount) {
                                K ek;
                                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;
                                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) {
                    if (binCount >= TREEIFY_THRESHOLD)
                        //把链表转成红黑树
                        treeifyBin(tab, i);
                    if (oldVal != null)
                        return oldVal;
                    break;
                }
            }
        }
        //这个addCount方法中会先统计当前map中的元素个数，然后判断是否需要扩容，
        //如果需要就会进行扩容
        addCount(1L, binCount);
        return null;
  }

四、扩容方法

扩容时首先涉及到计算map中总元素的个数,ConcurrentHashMap,和HashMap计算元素个数的方法不一样，可以先看一下它的size方法

public int size() {
        long n = sumCount();
        return ((n < 0L) ? 0 :
                (n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE :
                (int)n);
}

final long sumCount() {
      //这个baseCount是类属性，每次put时都会尝试先给这个值+1(cas)，如果尝试失败，当前put的线程就会
      //随机出一个CounterCell数组的下标,并给这个下标处的CounterCell.value进行+1(cas)
      //所以最终统计map中元素的个数时是把baseCount和CounterCell中每个元素的value加起来
        CounterCell[] as = counterCells; CounterCell a;
        long sum = baseCount;
        if (as != null) {
            for (int i = 0; i < as.length; ++i) {
                if ((a = as[i]) != null)
                    sum += a.value;
            }
        }
        return sum;
 }

jdk8中扩容的基本原理是每个参与扩容的线程会先确定自己转移那几个下标上的元素(在old数组中从右往左开始转移),当开始转移某个下标时就会用synchronized加锁，这样别的线程就不能往这个位置处put元素了(被阻塞住了)，只有当转移完成后别的线程才可以获取到这个node对象的锁并往这个链表或者红黑树上put元素。每转移完old数组上的一个下标,就会给这下标处放一个hash值是-1的元素ForwardingNode,别的线程往这个位置put元素时就会发现当前正在扩容，然后就会帮助扩容。