线程安全的并发容器1.8版本的ConcurrentHashMap原理分析（二十三）

r囧r小猫 2022-12-31 02:26 50阅读 0赞

线程安全的并发容器1.8版本的ConcurrentHashMap原理分析：

***在***  ***1.8***  ***下的实现，与1.7相比：***

改进一：取消  segments  字段，直接采用  transient volatile HashEntry<K,V>\[\] table 保存数据，采用  table  数组元素作为锁，从而实现了对缩小锁的粒度，进一 步减少并发冲突的概率，并大量使用了采用了 CAS + synchronized  来保证并发安

全性。

改进二：将原先  table  数组＋单向链表的数据结构，变更为  table  数组＋单 向链表＋红黑树的结构。对于 hash  表来说，最核心的能力在于将  key hash  之后 能均匀的分布在数组中。如果 hash  之后散列的很均匀，那么  table  数组中的每个 队列长度主要为 0  或者  1 。但实际情况并非总是如此理想，虽然 ConcurrentHashMap 类默认的加载因子为  0.75 ，但是在数据量过大或者运气不佳的情况下，还是会存在一些队列长度过长的情况，如果还是采用单向列表方式， 那么查询某个节点的时间复杂度为 O(n) ；因此，对于个数超过  8( 默认值 ) 的列表， jdk1.8 中采用了红黑树的结构，那么查询的时间复杂度可以降低到  O(logN) ，可 以改进性能。

使用  Node （ 1.7  为  Entry ） 作为链表的数据结点，仍然包含  key ， value ， hash 和  next  四个属性。 红黑树的情况使用的是  TreeNode （ **extends**  Node ）。 根据数组元素中，第一个结点数据类型是 Node  还是  TreeNode  可以判断 该位置下是链表还是红黑树。 用于判断是否需要将链表转换为红黑树的阈值

static final int TREEIFY_THRESHOLD = 8;

用于判断是否需要将红黑树转换为链表的阈值

static final int UNTREEIFY_THRESHOLD = 6;

1、 核心数据结构和属性

Node  是最核心的内部类，它包装了  key-value  键值对。

static class Node<K,V> implements 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;
            }
    
            public final K getKey()       { return key; }
            public final V getValue()     { return val; }
            public final int hashCode()   { return key.hashCode() ^ val.hashCode(); }
            public final String toString(){ return key + "=" + val; }
            public final V setValue(V value) {
                throw new UnsupportedOperationException();
            }
    
            public final boolean equals(Object o) {
                Object k, v, u; Entry<?,?> e;
                return ((o instanceof Map.Entry) &&
                        (k = (e = (Entry<?,?>)o).getKey()) != null &&
                        (v = e.getValue()) != null &&
                        (k == key || k.equals(key)) &&
                        (v == (u = val) || v.equals(u)));
            }
    
            /**
             * Virtualized support for map.get(); overridden in subclasses.
             */
            Node<K,V> find(int h, Object k) {
                Node<K,V> e = this;
                if (k != null) {
                    do {
                        K ek;
                        if (e.hash == h &&
                            ((ek = e.key) == k || (ek != null && k.equals(ek))))
                            return e;
                    } while ((e = e.next) != null);
                }
                return null;
            }
        }

定义基本和  1.7  中的  HashEntry  相同。而这个  map  本身所持有的也是一个Node 型的数组, 增加了一个 find  方法来用以辅助  map.get() 方法。其实就是遍历链表，子类

中会覆盖这个方法。

static class Segment<K,V> extends ReentrantLock implements Serializable {
            private static final long serialVersionUID = 2249069246763182397L;
            final float loadFactor;
            Segment(float lf) { this.loadFactor = lf; }
        }

在  map  中还定义了  Segment  这个类，不过只是为了向前兼容而已，不做过 多考虑

2、*TreeNode*

树节点类，另外一个核心的数据结构。当链表长度过长的时候，会转换为 TreeNode。

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;
            }
    
            Node<K,V> find(int h, Object k) {
                return findTreeNode(h, k, null);
            }
    
            /**
             * Returns the TreeNode (or null if not found) for the given key
             * starting at given root.
             */
            final TreeNode<K,V> findTreeNode(int h, Object k, Class<?> kc) {
                if (k != null) {
                    TreeNode<K,V> p = this;
                    do  {
                        int ph, dir; K pk; TreeNode<K,V> q;
                        TreeNode<K,V> pl = p.left, pr = p.right;
                        if ((ph = p.hash) > h)
                            p = pl;
                        else if (ph < h)
                            p = pr;
                        else if ((pk = p.key) == k || (pk != null && k.equals(pk)))
                            return p;
                        else if (pl == null)
                            p = pr;
                        else if (pr == null)
                            p = pl;
                        else if ((kc != null ||
                                  (kc = comparableClassFor(k)) != null) &&
                                 (dir = compareComparables(kc, k, pk)) != 0)
                            p = (dir < 0) ? pl : pr;
                        else if ((q = pr.findTreeNode(h, k, kc)) != null)
                            return q;
                        else
                            p = pl;
                    } while (p != null);
                }
                return null;
            }
        }

与  1.8  中  HashMap  不同点：

1 、它并不是直接转换为红黑树，而是把这些结点放在  TreeBin  对象中，由 TreeBin 完成对红黑树的包装。

2 、 TreeNode  在  ConcurrentHashMap  扩展自  Node  类，而并非  HashMap  中的 扩展自 LinkedHashMap.Entry<K,V> 类，也就是说  TreeNode  带有  next  指针。

3、*TreeBin*

负责  TreeNode  节点。它代替了  TreeNode  的根节点，也就是说在实际的 ConcurrentHashMap“数组”中，存放的是  TreeBin  对象，而不是  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

4、*特殊的* *ForwardingNode*

一个特殊的  Node  结点， hash  值为  \-1 ，其中存储  nextTable  的引用。有 table 发生扩容的时候， ForwardingNode  发挥作用，作为一个占位符放在  table 中表示当前结点为 null  或者已经被移动。

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;
            }
    
            Node<K,V> find(int h, Object k) {
                // loop to avoid arbitrarily deep recursion on forwarding nodes
                outer: for (Node<K,V>[] tab = nextTable;;) {
                    Node<K,V> e; int n;
                    if (k == null || tab == null || (n = tab.length) == 0 ||
                        (e = tabAt(tab, (n - 1) & h)) == null)
                        return null;
                    for (;;) {
                        int eh; K ek;
                        if ((eh = e.hash) == h &&
                            ((ek = e.key) == k || (ek != null && k.equals(ek))))
                            return e;
                        if (eh < 0) {
                            if (e instanceof ForwardingNode) {
                                tab = ((ForwardingNode<K,V>)e).nextTable;
                                continue outer;
                            }
                            else
                                return e.find(h, k);
                        }
                        if ((e = e.next) == null)
                            return null;
                    }
                }
            }
        }

5、*sizeCtl*

用来控制  table  的初始化和扩容操作。 负数代表正在进行初始化或扩容操作

  \-1  代表正在初始化

  \-N  表示有  N-1  个线程正在进行扩容操作

0  为默认值，代表当时的  table  还没有被初始化 正数表示初始化大小或 Map  中的元素达到这个数量时，需要进行扩容了。

6、 核心方法

/*利用硬件级别的原子操作，获得在i位置上的Node节点
        * Unsafe.getObjectVolatile可以直接获取指定内存的数据，
        * 保证了每次拿到数据都是最新的*/
        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操作设置i位置上的Node节点*/
        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);
        }
    
        /*利用硬件级别的原子操作，设置在i位置上的Node节点
        * Unsafe.putObjectVolatile可以直接设定指定内存的数据，
        * 保证了其他线程访问这个节点时一定可以看到最新的数据*/
        static final <K,V> void setTabAt(Node<K,V>[] tab, int i, Node<K,V> v) {
            U.putObjectVolatile(tab, ((long)i << ASHIFT) + ABASE, v);
        }

7、构造方法

public ConcurrentHashMap18(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;
        }

可以发现，在  new  出一个  map  的实例时，并不会创建其中的数组等等相关 的部件，只是进行简单的属性设置而已，同样的，table  的大小也被规定为必须 是 2  的乘方数。

真正的初始化在放在了是在向  ConcurrentHashMap  中插入元素的时候发生 的。如调用 put 、 computeIfAbsent 、 compute 、 merge  等方法的时候，调用时机 是检查 table==null 。

8、get 操作

get  方法比较简单，给定一个  key  来确定  value  的时候，必须满足两个条件 key 相同  hash  值相同，对于节点可能在链表或树上的情况，需要分别去查找。

public V get(Object key) {
            Node<K,V>[] tab; Node<K,V> e, p; int n, eh; K ek;
            int h = spread(key.hashCode());/*计算hash值*/
            /*根据hash值确定节点位置*/
            if ((tab = table) != null && (n = tab.length) > 0 &&
                (e = tabAt(tab, (n - 1) & h)) != null) {
                /*Node数组中的节点就是要找的节点*/
                if ((eh = e.hash) == h) {
                    if ((ek = e.key) == key || (ek != null && key.equals(ek)))
                        return e.val;
                }
                /*eh<0 说明这个节点在树上 调用树的find方法寻找*/
                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;
        }

9、put 操作

public V put(K key, V value) {
            return putVal(key, value, false);
        }
    
        /** 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());/*计算hash值*/
            int 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();/*如果table为空的话，初始化table*/
                else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
                    /*Node数组中的元素，这个位置没有值 ，使用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数组中的元素，这个位置是Hash冲突组成链表的头结点
                    * 或者是红黑树的根节点*/
                    synchronized (f) {
                        if (tabAt(tab, i) == f) {
                            /*fh>0 说明这个节点是一个链表的节点 不是树的节点*/
                            if (fh >= 0) {
                                binCount = 1;
                                for (Node<K,V> e = f;; ++binCount) {
                                    K ek;
                                    /*put操作和putIfAbsent操作业务实现*/
                                    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) {
                        /*达到临界值8 就需要把链表转换为树结构*/
                        if (binCount >= TREEIFY_THRESHOLD)
                            treeifyBin(tab, i);
                        if (oldVal != null)
                            return oldVal;
                        break;
                    }
                }
            }
            /*Map的元素数量+1，并检查是否需要扩容*/
            addCount(1L, binCount);
            return null;
        }

总结来说， put  方法就是，沿用  HashMap  的  put  方法的思想，根据  hash  值 计算这个新插入的点在 table  中的位置  i ，如果  i  位置是空的，直接放进去，否则 进行判断，如果 i  位置是树节点，按照树的方式插入新的节点，否则把  i  插入到 链表的末尾。整体流程上，就是首先定义不允许 key 或 value 为 null 的情况放入 对于每 一个放入的值，首先利用 spread  方法对  key  的  hashcode  进行一次  hash  计算，由 此来确定这个值在 table  中的位置。 如果这个位置是空的，那么直接放入，而且不需要加锁操作。 如果这个位置存在结点，说明发生了 hash  碰撞，首先判断这个节点的类型。 如果是链表节点, 则得到的结点就是  hash  值相同的节点组成的链表的头节点。需 要依次向后遍历确定这个新加入的值所在位置。如果遇到 hash  值与  key  值都与 新加入节点是一致的情况，则只需要更新 value  值即可。否则依次向后遍历，直 到链表尾插入这个结点。如果加入这个节点以后链表长度大于 8 ，就把这个链表 转换成红黑树。如果这个节点的类型已经是树节点的话，直接调用树节点的插入 方法进行插入新的值

10、初始化

前面说过，构造方法中并没有真正初始化，真正的初始化在放在了是在向 ConcurrentHashMap 中插入元素的时候发生的。具体实现的方法就是  initTable

点击 进入 tab = initTable();/*如果table为空的话，初始化table*/

private final Node<K,V>[] initTable() {
            Node<K,V>[] tab; int sc;
            while ((tab = table) == null || tab.length == 0) {
                /*小于0表示有其他线程正在进行初始化操作，把当前线程CPU时间让出来。
                因为对于table的初始化工作，只能有一个线程在进行。*/
                if ((sc = sizeCtl) < 0)
                    Thread.yield(); // lost initialization race; just spin
                /*利用CAS操作把sizectl的值置为-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右移2位本质上就是n变为n原值的1/4，所以
                            * sc=0.75*n */
                            sc = n - (n >>> 2);
                        }
                    } finally {
                        /*将设置成扩容的阈值*/
                        sizeCtl = sc;
                    }
                    break;
                }
            }
            return tab;
        }

11、transfer方法

当  ConcurrentHashMap  容量不足的时候，需要对  table  进行扩容。这个方法 的基本思想跟 HashMap  是很像的，但是由于它是支持并发扩容的，所以要复杂 的多。我们不深入源码去讲述，只讲述其大概原理。 为何要并发扩容？因为在扩容的时候，总是会涉及到从一个“数组”到另一 个“数组”拷贝的操作，如果这个操作能够并发进行，就能利用并发处理去减少 扩容带来的时间影响。整个扩容操作分为两个部分：

第一部分是构建一个  nextTable, 它的容量是原来的  2  倍。

private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
            int n = tab.length, stride;
            if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
                stride = MIN_TRANSFER_STRIDE; // subdivide range
            if (nextTab == null) {            // initiating
                try {
                    @SuppressWarnings("unchecked")//容量扩成原来的2 倍
                    Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
                    nextTab = nt;
                } catch (Throwable ex) {      // try to cope with OOME
                    sizeCtl = Integer.MAX_VALUE;
                    return;
                }
                nextTable = nextTab;
                transferIndex = n;
            }
            int nextn = nextTab.length;
            ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
            boolean advance = true;
            boolean finishing = false; // to ensure sweep before committing nextTab
            for (int i = 0, bound = 0;;) {
                Node<K,V> f; int fh;
                while (advance) {
                    int nextIndex, nextBound;
                    if (--i >= bound || finishing)
                        advance = false;
                    else if ((nextIndex = transferIndex) <= 0) {
                        i = -1;
                        advance = false;
                    }
                    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为新容量的0.75倍*/
                        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
                else {
                    synchronized (f) {
                        if (tabAt(tab, i) == f) {
                            Node<K,V> ln, hn;
                            if (fh >= 0) {
                                int runBit = fh & n;
                                Node<K,V> lastRun = f;
                                for (Node<K,V> p = f.next; p != null; p = p.next) {
                                    int b = p.hash & n;
                                    if (b != runBit) {
                                        runBit = b;
                                        lastRun = p;
                                    }
                                }
                                if (runBit == 0) {
                                    ln = lastRun;
                                    hn = null;
                                }
                                else {
                                    hn = lastRun;
                                    ln = null;
                                }
                                for (Node<K,V> p = f; p != lastRun; p = p.next) {
                                    int ph = p.hash; K pk = p.key; V pv = p.val;
                                    if ((ph & n) == 0)
                                        ln = new Node<K,V>(ph, pk, pv, ln);
                                    else
                                        hn = new Node<K,V>(ph, pk, pv, hn);
                                }
                                setTabAt(nextTab, i, ln);
                                setTabAt(nextTab, i + n, hn);
                                setTabAt(tab, i, fwd);
                                advance = true;
                            }
                            else if (f instanceof TreeBin) {
                                TreeBin<K,V> t = (TreeBin<K,V>)f;
                                TreeNode<K,V> lo = null, loTail = null;
                                TreeNode<K,V> hi = null, hiTail = null;
                                int lc = 0, hc = 0;
                                for (Node<K,V> e = t.first; e != null; e = e.next) {
                                    int h = e.hash;
                                    TreeNode<K,V> p = new TreeNode<K,V>
                                        (h, e.key, e.val, null, null);
                                    if ((h & n) == 0) {
                                        if ((p.prev = loTail) == null)
                                            lo = p;
                                        else
                                            loTail.next = p;
                                        loTail = p;
                                        ++lc;
                                    }
                                    else {
                                        if ((p.prev = hiTail) == null)
                                            hi = p;
                                        else
                                            hiTail.next = p;
                                        hiTail = p;
                                        ++hc;
                                    }
                                }
                                ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) :
                                    (hc != 0) ? new TreeBin<K,V>(lo) : t;
                                hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) :
                                    (lc != 0) ? new TreeBin<K,V>(hi) : t;
                                setTabAt(nextTab, i, ln);
                                setTabAt(nextTab, i + n, hn);
                                setTabAt(tab, i, fwd);
                                advance = true;
                            }
                        }
                    }
                }
            }
        }

第二个部分就是将原来  table  中的元素复制到  nextTable  中，这里允许多线程 进行操作。 整个扩容流程就是遍历和复制： 为 null  或者已经处理过的节点，会被设置为  forwardNode  节点，当线程准备 扩容时，发现节点是 forwardNode  节点，跳过这个节点，继续寻找未处理的节点， 找到了，对节点上锁， 如果这个位置是 Node  节点（ fh>=0 ），说明它是一个链表，就构造一个反序 链表，把他们分别放在 nextTable  的  i  和  i+n  的位置上 如果这个位置是 TreeBin  节点（ fh<0 ），也做一个反序处理，并且判断是否 需要红黑树转链表，把处理的结果分别放在 nextTable  的  i  和  i+n  的位置上 遍历过所有的节点以后就完成了复制工作，这时让 nextTable  作为新的  table ， 并且更新 sizeCtl  为新容量的  0.75  倍 ，完成扩容。 并发扩容其实就是将数据迁移任务拆分成多个小迁移任务，在实现上使用了 一个变量 stride  作为步长控制，每个线程每次负责迁移其中的一部分。

12、remove方法：

移除方法的基本流程和 put 方法很类似，只不过操作由插入数据变为移除数 据而已，而且如果存在红黑树的情况下，会检查是否需要将红黑树转为链表的步 骤。不再重复讲述。

13、treeifyBin 方法：

用于将过长的链表转换为  TreeBin  对象。但是他并不是直接转换，而是进行 一次容量判断，如果容量没有达到转换的要求，直接进行扩容操作并返回；如果 满足条件才将链表的结构转换为 TreeBin  ，这与  HashMap  不同的是，它并没有 把 TreeNode  直接放入红黑树，而是利用了  TreeBin  这个小容器来封装所有的 TreeNode。

14、size

在  JDK1.8  版本中，对于  size  的计算，在扩容和  addCount() 方法就已经有处理 了，可以注意一下 Put  函数，里面就有  addCount() 函数，早就计算好的，然后你 size 的时候直接给你。 JDK1.7  是在调用  size() 方法才去计算，其实在并发集合中去 计算 size  是没有多大的意义的，因为  size  是实时在变的。 在具体实现上，计算大小的核心方法都是 sumCount()

final long sumCount() {
            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;
        }

可以看见，统计数量时使用了  baseCount 、和  CounterCell  类型的变量 counterCells 。其实  baseCount  就是记录容器数量的，而  counterCells  则是记录 CAS 更新  baseCounter  值时，由于高并发而导致失败的值。这两个变量的变化在 addCount() 方法中有体现，大致的流程就是：

1 、对  baseCount  做  CAS  自增操作。

2 、如果并发导致  baseCount CAS  失败了，则使用  counterCells 。

3 、如果  counterCells CAS  失败了，在  fullAddCount  方法中，会继续死循环 操作，直到成功。