ConcurrentHashMap原理分析(一)-综述

概述

　　ConcurrentHashMap，一个线程安全的高性能集合，存储结构和HashMap一样，都是采用数组进行分桶，之后再每个桶中挂一个链表，当链表长度大于8的时候转为红黑树，其实现线程安全的基本原理是采用CAS + synchronized组合，当数组的桶中没有元素时采用CAS插入，相反，则采用synchronized加锁插入，除此之外在扩容和记录size方面也做了很多的优化，扩容允许多个线程共同协助扩容，而记录size的方式则采用类似LongAddr的方式，提高并发性，本片文章是介绍ConcurrentHashMap的第一篇，主要介绍下其结构，put()、get()方法，后面几篇文章会介绍其他方法。

ConcurrentHashMap存储结构

 从上图可以清晰的看到其存储结构是采用数组 + 链表 + 红黑树的结构，下面就介绍一下每一种存储结构在代码中的表现形式。

数组

    transient volatile Node<K,V>[] table;
    private transient volatile Node<K,V>[] nextTable;

可以看到数组中存的是Node，Node就是构成链表的节点。第二个nextTable是扩容之后的数组，在扩容的时候会使用。

链表

static class Node<K,V> implements Map.Entry<K,V> {
        final int hash;
        final K key;
        volatile V val;
        volatile Node<K,V> next;

        Node(int hash, K key, V val, Node<K,V> next) {
            this.hash = hash;
            this.key = key;
            this.val = val;
            this.next = next;
        }
//省略部分代码
}

一个典型的单链表存储结构，里面保存着key,val，以及这个key对应的hash值，next表示指向下一个Node。

红黑树

static final class TreeNode<K,V> extends Node<K,V> {
        TreeNode<K,V> parent;  // red-black tree links
        TreeNode<K,V> left;
        TreeNode<K,V> right;
        TreeNode<K,V> prev;    // needed to unlink next upon deletion
        boolean red;

        TreeNode(int hash, K key, V val, Node<K,V> next,
                 TreeNode<K,V> parent) {
            super(hash, key, val, next);
            this.parent = parent;
        }
//省略部分代码
}

TreeNode是构成红黑树的节点，其继承了Node节点，用于保存key,val,hash等值。但是在数组中并不直接保存TreeNode，一开始在没看源码之前，我以为数组中保存的是红黑树的根节点，其实不是，是下面这个东东。

static final class TreeBin<K,V> extends Node<K,V> {
        TreeNode<K,V> root;
        volatile TreeNode<K,V> first;
        volatile Thread waiter;
        volatile int lockState;
        // values for lockState
        static final int WRITER = 1; // set while holding write lock
        static final int WAITER = 2; // set when waiting for write lock
        static final int READER = 4; // increment value for setting read lock
//省略部分代码
)

这个类封装了TreeNode,而且提供了链表转红黑树，以及红黑树的增删改查方法。

其他节点

 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;
        }
//省略部分代码
}

这个节点正常情况下在ConcurrentHashMap中是不存在的，只有当扩容的时候才会存在，该节点中有一个nextTable字段，用于指向扩容之后的数组，其使用方法是这样的，扩容的时候需要把旧数组的数据拷贝到新数组，当某个桶中的数据被拷贝完成之后，就把旧数组的该桶标记为ForwardingNode，当别的线程访问到这个桶，发现被标记为ForwardingNode就知道该桶已经被copy到了新数组，之后就可以根据这个做相应的处理。

ConcurrentHashMap关键属性分析

这些属性先有个印象，都会在之后的源码中使用，不用现在就搞明白。

    //最大容量
    private static final int MAXIMUM_CAPACITY = 1 << 30;
    //默认初始化容量
    private static final int DEFAULT_CAPACITY = 16;
    //负载因子
    private static final float LOAD_FACTOR = 0.75f;
    //链表转为红黑树的临界值
    static final int TREEIFY_THRESHOLD = 8;
    //红黑树转为链表的临界值
    static final int UNTREEIFY_THRESHOLD = 6;
    //当容量大于64时，链表才会转为红黑树，否则，即便链表长度大于8，也不会转，而是会扩容
    static final int MIN_TREEIFY_CAPACITY = 64;
    //以上的几个属性和HashMap一模一样


    //扩容相关，每个线程负责最小桶个数
    private static final int MIN_TRANSFER_STRIDE = 16;
    //扩容相关，为了计算sizeCtl
    private static int RESIZE_STAMP_BITS = 16;
    //最大辅助扩容线程数量
    private static final int MAX_RESIZERS = (1 << (32 - RESIZE_STAMP_BITS)) - 1;
    //扩容相关，为了计算sizeCtl
    private static final int RESIZE_STAMP_SHIFT = 32 - RESIZE_STAMP_BITS;
    //下面几个是状态值
    //MOVED表示正在扩容
    static final int MOVED     = -1; // hash for forwarding nodes
    //-2表示红黑树标识
    static final int TREEBIN   = -2; // hash for roots of trees
    static final int RESERVED  = -3; // hash for transient reservations
    //计算Hash值使用
    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;
    //这个sizeCtl非常重要，基本上在代码中到处可以看到它的身影，后面会单独分析一下
    private transient volatile int sizeCtl;
    //扩容相关
    private transient volatile int transferIndex;
    //计算容器size相关
    private transient volatile int cellsBusy;
    //计算容器size相关，在介绍相关代码的时候详细介绍
    private transient volatile CounterCell[] counterCells;

上面的最开始的几个属性应该很好理解，后面的几个属性可能不知道有什么用，没关系，等到介绍相关代码的时候都会介绍的，这里着重介绍下sizeCtl,这个字段控制着扩容和table初始化，在不同的地方有不同的用处，下面列举一下其每个标识的意思：

  • 负数代表正在进行初始化或扩容操作
  • -1代表正在初始化
  • -N 表示，这个高16位表示当前扩容的标志，每次扩容都会生成一个不一样的标志，低16位表示参与扩容的线程数量
  • 正数或0，0代表hash表还没有被初始化，正数表示达到这个值需要扩容，其实就等于(容量 * 负载因子)

CAS操作

上面介绍了ConcurrentHashMap是通过CAS + synchronized保证线程安全的，那CAS操作有哪些，如下：

    
    //获取数组中对应索引的值
    static final <K,V> Node<K,V> tabAt(Node<K,V>[] tab, int i) {
        return (Node<K,V>)U.getObjectVolatile(tab, ((long)i << ASHIFT) + ABASE);
    }
　　　//修改数组对应索引的值，这个是真正的CAS操作
    static final <K,V> boolean casTabAt(Node<K,V>[] tab, int i,
                                        Node<K,V> c, Node<K,V> v) {
        return U.compareAndSwapObject(tab, ((long)i << ASHIFT) + ABASE, c, v);
    }
    //设置数组对应索引的值
    static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) {
        U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v);
    }

上面三个方法，我看很多文章把这三个方法都归类为CAS操作，其实第一个和第三个我觉得并不是，比如第一个方法，只是强制从主内存获取数据，第三个方法是修改完数据之后强制刷新到主内存，同时通知其他线程失效，只是为了保证可见性，而且这两个要求被修改的对象一定要被volatile修饰，这也是上面在介绍table的时候被volatile修饰的原因。

put()方法

put方法实际调用的是putVal()方法，下面分析下putVal方法。

 1 final V putVal(K key, V value, boolean onlyIfAbsent) {
 2         if (key == null || value == null) throw new NullPointerException();
 3         //这个计算hash值的方法和hashMap不同
 4         int hash = spread(key.hashCode());
 5         //记录链表节点个数
 6         int binCount = 0;
 7         //这个死循环的作用是为了保证CAS一定可以成功，否则就一直重试
 8         for (Node<K,V>[] tab = table;;) {
 9             Node<K,V> f; int n, i, fh;
10             //如果table还没有初始化，初始化
11             if (tab == null || (n = tab.length) == 0)
12                 //初始化数组，后面会分析，说明1
13                 tab = initTable();
14             //如果通过hash值定位到桶的位置为null，直接通过CAS插入，上面死循环就是为了这里
15             else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
16                 if (casTabAt(tab, i, null,
17                              new Node<K,V>(hash, key, value, null)))
18                     break;                   // no lock when adding to empty bin
19             }
20             //如果发现节点的Hash值为MOVED，协助扩容，至于为什么hash值会为MOVEN，后面会说明，说明2
21             else if ((fh = f.hash) == MOVED)
22                 //协助扩容，在讲解扩容的时候再讲解
23                 tab = helpTransfer(tab, f);
24             else {
25                 //到这里说明桶中有值
26                 V oldVal = null;
27                 //不管是链表还是红黑树都加锁处理，防止别的线程修改
28                 synchronized (f) {
29                     //这里直接从主内存重新获取，双重检验，防止已经被别的线程修改了
30                     if (tabAt(tab, i) == f) {
31                         //fh >= 0，说明是链表，为什么fh>=0就是链表，这个就是hash值计算的神奇的地方，所有的key的hash都是大于等于0的，
32                         //红黑树的hash值为-2，至于为什么为-2后面会说明，说明3
33                         if (fh >= 0) {
34                             //这里就开始记录链表中节点个数了，为了转为红黑树做好记录
35                             binCount = 1;
36                             //for循环遍历链表
37                             for (Node<K,V> e = f;; ++binCount) {
38                                 K ek;
39                                 //如果key相同，就替换value
40                                 if (e.hash == hash &&
41                                     ((ek = e.key) == key ||
42                                      (ek != null && key.equals(ek)))) {
43                                     oldVal = e.val;
44                                     //这个参数传的是false
45                                     if (!onlyIfAbsent)
46                                         e.val = value;
47                                     break;
48                                 }
49                                 //遍历没有发现有相同key的，就挂在链表的末尾
50                                 Node<K,V> pred = e;
51                                 if ((e = e.next) == null) {
52                                     pred.next = new Node<K,V>(hash, key,
53                                                               value, null);
54                                     break;
55                                 }
56                             }
57                         }
58                         //如果是红黑树，这里就是上面介绍的，数组中存的不是TreeNode,而是TreeBin
59                         else if (f instanceof TreeBin) {
60                             Node<K,V> p;
61                             binCount = 2;
62                             //向红黑树插入
63                             if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
64                                                            value)) != null) {
65                                 oldVal = p.val;
66                                 if (!onlyIfAbsent)
67                                     p.val = value;
68                             }
69                         }
70                     }
71                 }
72                 if (binCount != 0) {
73                     //如果链表长度大于等于8，转为红黑树,至于怎么转在介绍红黑树部分的时候再详细说
74                     if (binCount >= TREEIFY_THRESHOLD)
75                         treeifyBin(tab, i);
76                     if (oldVal != null)
77                         return oldVal;
78                     break;
79                 }
80             }
81         }
82         //计算size++,不过是线程安全的方式，这里这篇文章先不介绍，之后会专门介绍
83         addCount(1L, binCount);
84         return null;
85     }

整个过程梳理如下：

  1. 数组没有初始化就先初始化数组
  2. 计算当前插入的key的hash值
  3. 根据第二步的hash值定位到桶的位置，如果为null，直接CAS自旋插入
  4. 如果是链表就遍历链表，有相同的key就替换，没有就插入到链表尾部
  5. 如果是红黑树直接插入
  6. 判断链表长度是否超过8，超过就转为红黑树
  7. ConcurrentHashMap元素个数加1

上面代码中标红的地方说明：

说明一：initTable()

private final Node<K,V>[] initTable() {
        Node<K,V>[] tab; int sc;
        while ((tab = table) == null || tab.length == 0) {
            //如果这个值小于零，说明有别的线程在初始化
            if ((sc = sizeCtl) < 0)
                //让出CPU时间，注意这时线程依然是RUNNABLE状态
                //这里使用yield没有风险，因为即便这个线程又竞争到CPU，再次循环到这里它还会让出CPU的
                Thread.yield(); // lost initialization race; just spin
            //初始状态SIZECTL为0，通过CAS修改为-1
            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;
                        //扩容点，比如n = 16,最后计算出来的sc = 12
                        sc = n - (n >>> 2);
                    }
                } finally {
                    sizeCtl = sc;
                }
                break;
            }
        }
        return tab;
    }

说明二：扩容状态为什么hash为MOVEN

//构造方法，里面使用super，也就是他的父类Node的构造方法   
 ForwardingNode(Node<K,V>[] tab) {
            super(MOVED, null, null, null);
            this.nextTable = tab;
        }

上面介绍ForwardingNode的时候说过，这个是扩容的时候，如果这个桶处理过了就设置为该节点，这个类的构造方法可以看出，它会把hash值设置为MOVEN状态。

说明三：红黑树TreeBin的hash值为什么为-2

TreeBin(TreeNode<K,V> b) {
            super(TREEBIN, null, null, null);
            this.first = b;
            TreeNode<K,V> r = null;
            for (TreeNode<K,V> x = b, next; x != null; x = next) {
                next = (TreeNode<K,V>)x.next;
                x.left = x.right = null;
                if (r == null) {
                    x.parent = null;
                    x.red = false;
                    r = x;
                }
//省略部分代码
}

这个是TreeBin的构造方法，这个super同样是Node的构造方法，hash值为TREEBIN = -2

get()方法

 public V get(Object key) {
        Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
        //计算key的hash值
        int h = spread(key.hashCode());
        //数组不为空，获取对应桶的值
        if ((tab = table) != null && (n = tab.length) > 0 &&
            (e = tabAt(tab, (n - 1) & h)) != null) {
            //获取到，直接返回value
            if ((eh = e.hash) == h) {
                if ((ek = e.key) == key || (ek != null && key.equals(ek)))
                    return e.val;
            }
            //小于0，就是上面介绍的TREEBIN状态，是红黑树，在红黑树中查找
            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;
    }

get方法很简单，就是去各个数据结构中找，不过红黑树的遍历还是要好好看看的，这里先不分析，红黑树这玩意为了实现自平衡，定义了很多的限制条件，实现起来的复杂度真是爆炸，之后文章会分析，不过代码看的我都快吐了，哈哈哈。

　　　　

 总结

    本篇文章就先分析到这，不然就太长了，本文介绍了ConcurrentHashMap的存储结构，节点构成，以及初始化方法，put和get方法，整体来说这部分比较简单，ConcurrentHashMap复杂的部分是扩容和计数，当然我自己觉得红黑树部分是最复杂的，后面再慢慢介绍。

【推广】 免费学中医，健康全家人
原文地址：https://www.cnblogs.com/gunduzi/p/13649860.html