ConcurrentHashMap 源码分析,基于JDK1.8

时间:2021-12-05 17:18:54

1:几个重要的成员变量:

private static final int MAXIMUM_CAPACITY = 1 << 30; //map 容器的最大容量

private static final int DEFAULT_CAPACITY = 16; // map容器的默认大小

private static final float LOAD_FACTOR = 0.75f;  //加载因子

static final int TREEIFY_THRESHOLD = 8;  //由链表转为树状结构的链表长度

static final int UNTREEIFY_THRESHOLD = 6; //由树状结构转为链表

static final int MIN_TREEIFY_CAPACITY = 64; //数组长度最小为64才会转为红黑树

 

transient volatile Node<K,V>[] table;  //Node数组 用于存储元素

private transient volatile Node<K,V>[] nextTable; //当扩容的时候用于临时存储数组链表

private transient volatile long baseCount; //保存着哈希表所有节点的个数总和,相当于hash      Map  size

private transient volatile int sizeCtl;

下面对sizeCtl这个属性进行说明:

  • 0:默认值
  • -1:代表哈希表正在进行初始化
  • 大于0:相当于 HashMap 中的 threshold,表示阈值
  • 小于-1:代表有多个线程正在进行扩容

由这些成员变量可以看到内部的数据结构是基于 数组+单链表 的数据结构的,其中链表会转化为树状结构;

2:构造方法

// 初始化 sizeCtl 参数

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;

}

3:几个重要的方法介绍 put  get remove三个方法的介绍;

3.1:put 方法:

public V put(K key, V value) {  //key=name1   value=aa

        return putVal(key, value, false);

}

接下来看看putVal这个方法的具体实现:分为4部分来解析:

 

 

 

 

第一部分:

final V putVal(K key, V value, boolean onlyIfAbsent) { //key=name1  value=aa onlyIfAbsent=false

        if (key == null || value == null) throw new NullPointerException();

        int hash = spread(key.hashCode());  //获取到key的hash值

        int binCount = 0;  // 赋值 binCount=0

        for (Node<K,V>[] tab = table;;) {  //进入循环方法体

            Node<K,V> f; int n, i, fh;

            if (tab == null || (n = tab.length) == 0)  //如果是添加第一个元素进入扩容方法

                tab = initTable();

            else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {

                if (casTabAt(tab, i, null,

                             new Node<K,V>(hash, key, value, null)))

                    break;                   // no lock when adding to empty bin

            }

 

//接下来看看扩容方法的实现:

private final Node<K,V>[] initTable() {

        Node<K,V>[] tab; int sc;

        while ((tab = table) == null || tab.length == 0) {

            if ((sc = sizeCtl) < 0)  //当有其他线程在进行扩容 当前线程yield 放弃cpu使用

                Thread.yield(); // lost initialization race; just spin

            else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {//通过CAS将sc更新为 -1

                try {

//这里假设是第一次put操作时进行扩容   sc=-1

                    if ((tab = table) == null || tab.length == 0) {

                        int n = (sc > 0) ? sc : DEFAULT_CAPACITY; //n=16

                        @SuppressWarnings("unchecked")

                        Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n]; //新建一个长度为16的Node数组

                        table = tab = nt;  // 将nt赋值给成员变量table

                        sc = n - (n >>> 2); // sc=12

                    }

                } finally {

                    sizeCtl = sc;  // sizeCtl=-1  这里保证当其他线程进入initTable方法时线程会被 yield;

                }

                break;

            }

        }

        return tab;

    }

第二部分:

接下来看看put方法的第二部分操作:

else if ((fh = f.hash) == MOVED)  // MOVED=-1

                tab = helpTransfer(tab, f);

简单说明下上面的方法:当 数组中的链表正在迁移中(如:扩容的情况下),则进入该方法;表明帮助迁移

第三部分:

这一部分虽然看起来代码量比较大,但是具体的操作和hashmap类似,主要的区别是

 对这个链表进行加锁后处理,synchronized对象锁后进行相应的操作。tabAt保证获取到的对象是可见性的;其他的操作和hashmap中是一致的;就不做过多的说明了;

V oldVal = null;

                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;

                }

            }

接下来看看第四部分的代码:在put 方法中只有一句,如下所示:

addCount(1L, binCount); //这里的 binCount是指对操作那个Node链表的长度

每次添加Node元素到链表中binCount都会加1

接下来主要是分析addCount 里面的方法:1:更新baseCount  2:判断是否需要扩容

private final void addCount(long x, int check) {  //x=1   check=4

        CounterCell[] as; long b, s;

        if ((as = counterCells) != null ||

            !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) {//将baseCount 更新为b + x 即总量+1

            CounterCell a; long v; int m;

            boolean uncontended = 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();

        }

        if (check >= 0) {  // 链表长度大于0

            Node<K,V>[] tab, nt; int n, sc;

// sizeCtl=-1  s=0  tab!=null  n为node[] 数组的长度

            while (s >= (long)(sc = sizeCtl) && (tab = table) != null &&

                   (n = tab.length) < MAXIMUM_CAPACITY) {

                int rs = 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))  //sc更新为1

                        transfer(tab, nt); 

                }

                else if (U.compareAndSwapInt(this, SIZECTL, sc,

                                             (rs << RESIZE_STAMP_SHIFT) + 2))

                    transfer(tab, null);  //进入扩容的方法

                s = sumCount();

            }

        }

    }

 

下面看看 transfer的方法进入扩容的操作:

这个方法比较复杂,分为3个部分进行分析:

//这里tab为16的Node数组  nextTab为 null

private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {

        int n = tab.length, stride;

//计算单个线程允许处理的最少table桶首节点个数,不能小于 16

        if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)

            stride = MIN_TRANSFER_STRIDE; // subdivide range

//刚开始扩容  初始化nextTab

        if (nextTab == null) {            // initiating

            try {

                @SuppressWarnings("unchecked")

                Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1]; //创建长度为32的Node数组

                nextTab = nt;  // nextTab为32的空数组

            } catch (Throwable ex) {      // try to cope with OOME

                sizeCtl = Integer.MAX_VALUE;

                return;

            }

            nextTable = nextTab;  // nextTable变量为长度32的数组

            transferIndex = n;  // transferIndex=32  指向数组的最后一个桶

        }

        int nextn = nextTab.length;

//定义ForwardingNode用于标记已经迁移完的桶

        ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);

 

 

 

 

 

 

 

 

 

接下来看看扩容的第二部分:

boolean advance = true;

boolean finishing = false; // to ensure sweep before committing nextTab

// bound 表示当前线程需要处理桶节点区间的下限

        for (int i = 0, bound = 0;;) {  //进入for循环,这里没有结束条件

            Node<K,V> f; int fh;

            while (advance) {

                int nextIndex, nextBound;

                if (--i >= bound || finishing)

                    advance = false;

                else if ((nextIndex = transferIndex) <= 0) { // 这里 nextIndex=32 如果nextIndex<=0 则说明没有需要迁移的桶

                    i = -1;

                    advance = false;

                }

//更新transferIndex

//为当前线程分配任务,处理桶区间为(nextBound, nextIndex)

                else if (U.compareAndSwapInt

                         (this, TRANSFERINDEX, nextIndex,

                          nextBound = (nextIndex > stride ?

                                       nextIndex - stride : 0))) {

                    bound = nextBound;

                    i = nextIndex - 1;

                    advance = false;

                }

            }

            if (i < 0 || i >= n || i + n >= nextn) {

                int sc;

                if (finishing) {

                    nextTable = null;

                    table = nextTab;

                    sizeCtl = (n << 1) - (n >>> 1);

                    return;

                }

                if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {

                    if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)

                        return;

                    finishing = advance = true;

                    i = n; // recheck before commit

                }

            }

            else if ((f = tabAt(tab, i)) == null)

                advance = casTabAt(tab, i, null, fwd);

            else if ((fh = f.hash) == MOVED)

                advance = true; // already processed