ConcurrentHashMap(JDK1.8)put分析（一）

ConcurrentHashMap和HashMap的put值过程有些类似，ConcurrentHashMap的结构也是table + 链表+ 红黑树；在put值时，锁粒度是table的元素；也就是说，当put值时定位到table的第 i 个元素，那么就会给table[i]上锁；其他线程在put值时也定位到 i 时就需要等待获取锁；如果是其他位置则不需要等待锁，可以进行put操作。
下面具体分析ConcurrentHashMap的put值过程

1.计算hash值

ConcurrentHashMap在put值的时候会计算key的hash值，和HashMap类似；在ConcurrentHashMap计算hash值的是spread方法

    public V put(K key, V value) {
        return putVal(key, value, false);
    }
			||
			||
			/
   final V putVal(K key, V value, boolean onlyIfAbsent) {
        if (key == null || value == null) throw new NullPointerException();
        int hash = spread(key.hashCode());
        ..
        ..
        ..
        }
        
			||
			||
			/

    static final int spread(int h) {
        return (h ^ (h >>> 16)) & HASH_BITS;
    }
    HASH_BITS = 0x7fffffff;(最大整数)

获取key的hashcode
利用hashcode的高16位与低16位做" ^ "位运算，让高位也参与位运算，增加离散型。
再与HASH_BITS位运算，这是int最大正整数:最高位为0，其余位置为1，用来与hash值做 " & "运算；得到的结果保证最高位为0，其余位置不变；这样就保证得到的hash值是一个非负数；
举个例子：

hashcode:
h:
  10111101 00001101 01011111 00110101
h>>>16:
  00000000 00000000 10111101 00001101
h ^ h>>>16:
  10111101 00001101 11100010 00111000 (是一个负数)
& 
  01111111 11111111	11111111 11111111  [HASH_BITS(0x7fffffff)]
res  
  00111101 00001101 11100010 00111000 [最高位 1 -> 0;由负数变成正数]

2.put值

在put值时有几个判断，与HashMap类似

判断table是否已经初始化了；如果是没有初始化，就会先初始化table，再往table中添加值；
根据hash值，计算出key-value在table的位置i，判断table[i]是否已经插入了值；如果没有插入值，就将该key-value插入到table[i]中；
判断table[i]的hash值是否是MOVE,如果是MOVE表示正在扩容，需要该线程帮忙将旧容器的值移动到新容器中
将key-value插入到链表或者红黑树中
判断是否需要扩容

2.1初始化table

因为是多线程环境，就需要考虑到如果有多个线程同时初始化table的情况；假如现在有三个线程：A,B,C同时判断到table == null。这个时候三个线程都会同时试图来初始化table，如果一个线程抢先修改了sizeCtl，他就可以初始化table，其余线程只需要等待初始化完成即可；

    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[] tab = table;;) {
            Node f; int n, i, fh;
            if (tab == null || (n = tab.length) == 0)
                tab = initTable();
                ..
                ..
                ..
                ||
                ||
                /

    
    private final Node[] initTable() {
        Node[] tab; int sc;
        while ((tab = table) == null || tab.length == 0) {
            if ((sc = sizeCtl) < 0)//有线程在初始化table，就让出cpu，等待table初始化就行；
                Thread.yield(); // lost initialization race; just spin
            else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {//修改sizeCtl的值，sizeCtl=-1 
                try {
                    if ((tab = table) == null || tab.length == 0) {
                        int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
                        @SuppressWarnings("unchecked")
                        Node[] nt = (Node[])new Node[n];
                        table = tab = nt;
                        sc = n - (n >>> 2);
                    }
                } finally {
                    sizeCtl = sc;
                }
                break;
            }
        }
        return tab;
    }

1.sizeCtl判断是否已经有线程初始化table，如果sizeCtl=-1，表示已经有线程对table初始化；这个时候会调用yield()让出cpu执行权
2.通过CAS修改sizeCtl=-1，初始化table

这段代码，有意思的是进入到初始化table的分支时try{}finally{}代码块还要判断table==null，为什么呢？
可能是对应这样一个场景：现在有 A ，B，C三个线程都判断table=null进入到initTable，A，B线程先获取到sizeCtl =0，此时A抢先修改到sizeCtl =-1开始初始化，而B线程修改失败，再次通过循环获取sizeCtl =-1，调用yield()放弃cpu执行权，等待A线程初始化table；A线程初始化完成之后修改sizeCtl 的值，修改后的sizeCtl >0；

而C线程比A，B线程慢一点，当C线程获取到sizeCtl的值时，A线程已经完成了table的初始化sizeCtl >0，C线程获取到的sc=sizeCtl >0,因此不会进入到休眠状态，会尝试修改sizeCtl 的值，这个时候没有其他线程竞争修改值，因此会修改成功；又会将sizeCtl 的值修改为 -1 ，【此时：sc=A线程初始化之后sizeCtl 的值；sizeCtl =-1】

进入到try代码块，判断table!=null,不会再重新初始化table，进入finally块，sizeCtl =sc；将sizeCtl 的值还原到A线程初始化时候的值；

添加的table=null的判断保证了只会有一个线程能初始化table；

2.2思考

在try代码块中判断table是否已经初始化保证了只有一个线程能初始化table，这个和DCL(双重检测枷锁，差不多)；这段代码能修改吗？：

			 if ((sc = sizeCtl) < 0)
                Thread.yield();
            	    ||
            	    ||
            	    ||
            	    /
            if ((sc = sizeCtl) != 0)
            	Thread.yield();

因为只要sizeCtl != 0,就说明已经有线程在初始化table,其余线程都可以等待带线程初始化完成，这样当线程获初始化table的时就不用再判断table是否为null了，因为只能有一个线程能进入这个分支，对table进行初始化；
这样看起来似乎没什么问题，但是忽略了一个问题，在ConcurrentHashMap的构造对象中有一个构造方法可以指定初始容量，而保存初始容量的变量就是sizeCtl ，也就是说如果指定了初始容量sizeCtl 值就大于0；而指定初始容量时并不会初始化一个数组；因此不能修改成：(sc = sizeCtl) != 0，如果修改成这样，那么在指定初始容量之后，所有线程都不能初始化table了；
同时也可以看到在初始化时利用了sizeCtl的值：

 			try {
                    if ((tab = table) == null || tab.length == 0) {
                    //sc是sizeCtl修改之前的值，如果指定了容量，那么值就保存在sizeCtl中；
                    //这里的(sc > 0),这个判断其实就是判断在创建ConcurrentHashMap时有没有指定初始容量；指定了就用指定的容量；
                    //没有指定容量就用默认值:DEFAULT_CAPACITY
                        int n = (sc > 0) ? sc : DEFAULT_CAPACITY;//DEFAULT_CAPACITY=16;
                        @SuppressWarnings("unchecked")
                        Node[] nt = (Node[])new Node[n];
                        table = tab = nt;
                        sc = n - (n >>> 2);//sc = (3/4)*n
                    }
                } finally {
                    sizeCtl = sc;//sizeCtl 扩容阈值 sizeCtl = 0.75 * n
                }
                break;

3.通过hash值定位key-value位置

else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
                if (casTabAt(tab, i, null,
                             new Node(hash, key, value, null)))
                    break;                   // no lock when adding to empty bin
            }

这个分支其实就是判断table[i] = null，就将key-value值包装程成Node对象，放到table[i]中即可；

i = (n - 1) & hash；与HashMap一样，定位存储下标；
casTabAt（）将node添加到table中，添加成功：返回true，进入分支break，结束put操作；不成功：进入下一次循环，继续put操作直到成功为止；

4.判断ConcurrentHashMap是否正在扩容

            else if ((fh = f.hash) == MOVED)
                tab = helpTransfer(tab, f);

在上一个分支，如果table[i] != null,就会在接下来的分支中，首先通过node节点的hash值判断，table是否处于扩容状态，如果是扩容状态: hash == MOVE;
当判定在扩容时，会要求这个线程帮忙完成扩容:

    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;
    }

扩容暂时在这里不分析，在后面会讲到；因为扩容是添加node导致，在添加元素之后会判断是否需要扩容；在这里解释就会产生很奇怪的点：为什么会有node 的hash == MOVE（MOVE=-1）在这里不好解释；只需要了解，如果HASH = MOVE，就表示正在扩容，需要该线程协助扩容；

5. 判断节点的类型，并添加节点；

判断节点类型，并添加值：

如果是链表，就将值打包成Node，添加到链表中
如果是红黑树，就将值打包成TreeNode添加到红黑树中；

这个过程和 HashMap一样；唯一的区别是，在多线程环境下，在确定table的下标之后，会获取table[i]对象的锁，只能有一个线程在table[i]所在的链表或者红黑树put值；其他线程要在table[i]中put值需要等待获取锁；

	//	f = tabAt(tab, i = (n - 1) & hash)
		
            else {
                V oldVal = null;
                synchronized (f) {//获取table[i]的对象锁
                    if (tabAt(tab, i) == f) {//判断是否被修改
                        if (fh >= 0) {//根据node的hash值判断是链表还是红黑树
                            binCount = 1;
                            for (Node e = f;; ++binCount) {
                                K ek;
                               // 已经存在key，更新val，返回旧value
                                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;
                           // key在链表中不存在，将key-value打包成Node插入到链表表尾
                                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;//p !=null ,说明在红黑树中已经存在key，只是更新了value；
                                if (!onlyIfAbsent)
                                    p.val = value;
                            }
                        }
                    }
                }
                if (binCount != 0) {
                    if (binCount >= TREEIFY_THRESHOLD)
                        treeifyBin(tab, i);
                    if (oldVal != null)
                        return oldVal;
                    break;
                }
            }
        }
		
        addCount(1L, binCount);

为了更好的看清上面代码的整体逻辑，去掉部分代码：

			synchronized (f) {
			//判断节点没有变动；如果变动了，就计入下一次循环；
			//f发生改变的几种情况：
			//1:有可能链表变成红黑树，
			//2:可能容器在扩容；
			//3:如果其他线程在之前进行了remove操作，导致f被删除，这种情况也不能直接put；
			//还有一种情况：就是key-value的值被修改，这种情况对下一个线程put值没有影响；
			//因此可以看到在很多地方，对节点进行操作前，都会先判断节点有没有改变；
				if (tabAt(tab, i) == f) {
					if(fh>=0){
						binCount=1;
						链表，会将新node添加到链表末尾，在这个过程中binCount会记录链表的长度，用来判断是否需要将链表修改为红黑树；
						还有一种情况是这个key已经存在，就直接更新value；
					}else if(f instanceof TreeBin){
						binCount=2;
						红黑树，这里的binCount就是只用来表示该线程抢到锁，已经put值了；
					}
				}
			}

				if (binCount != 0) {//说明已经添加了node；binCount=0是因为线程没拿到锁，没有put值
                    if (binCount >= TREEIFY_THRESHOLD)//这里是用来判断链表需不需要改成红黑树;
                        treeifyBin(tab, i);
                    if (oldVal != null)//key已经存在了，会将value更新；并返回旧值oldValue
                        return oldVal;
                    break;
                }
                
		更新size，并且判断是否需要扩容；
		addCount(1L, binCount);

5.1为什么用hash值的正负数来判定是链表还是红黑树

上面有个地方最迷惑的是，用hash值的正负数来判断是链表还是红黑树；因为在之前我们提到过ConcurrentHashMap的hash值一定是不为负数的；我们先看下怎么将链表转换成红黑树的？treeifyBin(tab, i)方法是将链表转换成红黑树的，源码：

    private final void treeifyBin(Node[] tab, int index) {
        Node b; int n, sc;
        if (tab != null) {
        //判断table的长度是否够大，如果table长度没有64，就不用将链表 -> 红黑树；直接扩大数组长度；
            if ((n = tab.length) < MIN_TREEIFY_CAPACITY)
                tryPresize(n << 1);
            else if ((b = tabAt(tab, index)) != null && b.hash >= 0) {//将Node -> TreeNode
                synchronized (b) {
                    if (tabAt(tab, index) == b) {//保证只有一个线程转换Node；如果有线程已经转换了Node，其余线程则不必转换了；
                        TreeNode hd = null, tl = null;
                        //通过循环，将整个链表的Node节点，转换成TreeNode节点；
                        for (Node e = b; e != null; e = e.next) {
                            TreeNode p =
                                new TreeNode(e.hash, e.key, e.val,
                                                  null, null);
                            if ((p.prev = tl) == null)
                                hd = p;
                            else
                                tl.next = p;
                            tl = p;
                        }
                        //关键：链表的表头，封装成TreeBin；
                        setTabAt(tab, index, new TreeBin(hd));
                    }
                }
            }
        }
    }
    						||
    						||
    						/
    						
		 TreeBin(TreeNode b) {
		 //Node(int hash, K key, V val, Node next) {}
	            super(TREEBIN, null, null, null);
	            this.first = b;
	            TreeNode r = null;
				...
				...
				将TreeNode 链表 转换成 红黑树；
	    }
    						||
    						||
    						/
	    			
	    static final int TREEBIN   = -2; // hash for roots of trees

          通过上面的代码，发现treeifyBin只是将node -> TreeNode,真正将链表转换成红黑树的还是在创建TreeBin对象的时候；
          回到我们关注的问题：为什么用hash值的正负数来判定是链表还是红黑树？
我们已经知道在将链表转换成红黑树时，会创建一个TreeBin对象，并将这个TreeBin对象设置到table中，fh => table[i].hash；因此，只需要看TreeBin的hash就行了；在创建TreeBin对象时，使用的hash是：TREEBIn (-2) ,是一个负数；而链表的表头（table的元素table[i]）是直接创建的Node对象，使用的hash值也是利用key计算的hash因此是非负数;因此可以通过table元素的hash值正负数来判断是链表还是红黑树；
          在TreeBin对象中，保存了链表转换成红黑树之前的表头；和转换成红黑树之后生成的root节点；

   static final class TreeBin extends Node {
        TreeNode root;红黑树的根节点
        volatile TreeNode first;//在链表转换成红黑树之前，链表的表头；
        ...
        ...
        ...
        }

6.addCount(1L, binCount);

这段代码在多线程环境下计数，并判断是否需要扩容。这部分代码实在太多，首先是在多线程环境下如何计数？再一个就是多线程如何扩容？有点头痛，下一篇文章再分析吧。

ConcurrentHashMap(JDK1.8)put分析（一）

Java相关栏目本月热门文章