一：简述
本文基于jdk1.8对concurrentHashMap的源码进行分析，以put()方法为入口对concurrentHashMap的扩容机制，size计算方式等代码进行分析
二：concurrentHashMap成员变量
代码语言：javascript复制//最大容量
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;
//将链表转化为红黑树的node数组长度阈值
static final int MIN_TREEIFY_CAPACITY = 64;
//默认的线程迁移数据范围
private static final int MIN_TRANSFER_STRIDE = 16;
//当前服务器cpu数量
static final int NCPU = Runtime.getRuntime().availableProcessors();
//真正存储数据的容器
transient volatile Node<K,V>[] table;
//用于扩容的新数组
private transient volatile Node<K,V>[] nextTable;
//和counterCells一起用于计算concurrentHashMap的size
private transient volatile long baseCount;
// -1的时候代表正在node数组正在初始化 初始化之后赋值为扩容的阈值
private transient volatile int sizeCtl;
//数据迁移的索引
private transient volatile int transferIndex;
//用于计算concurrentHashMap的size时需要加cas锁的标记
private transient volatile int cellsBusy;
//用于计算concurrentHashMap的size 默认长度是2
private transient volatile CounterCell[] counterCells;三：concurrentHashMap的源码分析
以put方法为入口对源码进行分析
1.put方法流程图
2.源码分析
put()方法会调用putVal()方法，如果当前数组没有初始化那么会先调用initTable()方法初始化数组，然后根据计算好的数组下标查看当前下标下是否为null,如果是null,那么利用cas保证线程安全直接进行替换，如果不是null，那么需要解决hash冲突的问题，分链表和红黑树两种情况分别进行处理。
a.链表：遍历链表，如果有相同的key 进行覆盖的操作 否则添加到链表的尾部（尾插法）。
b.红黑树：遍历红黑树，如果有相同的key，进行覆盖操作，如果没有，那么构建红黑树的节点添加到红黑树，并且通过左旋或者右旋保证红黑树的平衡。
添加完元素之后判断链表的长度是否大于等于8，大于8 那么会调用treeifyBin()方法。
最后调用addcount()计算数组的总元素个数。
代码语言：javascript复制final V putVal(K key, V value, boolean onlyIfAbsent) {
//键和值都不能为空 否则抛出空指针异常
if (key == null || value == null) throw new NullPointerException();
//计算key的hash值
int hash = spread(key.hashCode());
int binCount = 0;
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh; K fk; V fv;
//如果tab 为空 那么需要先初始化数组
if (tab == null || (n = tab.length) == 0)
tab = initTable();
//(n - 1) & hash 计算当前的值应该存放的数组下标
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) {
//如果计算出的数组位置的node为null 直接用cas进行替换即可
if (casTabAt(tab, i, null, new Node<K,V>(hash, key, value)))
break; // no lock when adding to empty bin
}
//MOVED 代表当前节点正在进行数据迁移 那么直接去协助扩容
else if ((fh = f.hash) == MOVED)
tab = helpTransfer(tab, f);
//onlyIfAbsent true表示不能覆盖原有的值 默认是false
else if (onlyIfAbsent // check first node without acquiring lock
&& fh == hash
&& ((fk = f.key) == key || (fk != null && key.equals(fk)))
&& (fv = f.val) != null)
//如果存在相同的key 并且value不为空 直接返回已存在的值
return fv;
else {
//这段代码是处理存在hash冲突的情况的逻辑
V oldVal = null;
//针对node进行加锁
synchronized (f) {
if (tabAt(tab, i) == f) {
//针对链表的处理
if (fh >= 0) {
binCount = 1;
for (Node<K,V> e = f;; binCount) {
K ek;
//遍历链表 如果存在有相同的key 而且允许被覆盖 那么直接覆盖原有的值
//不需要cas来保证线程安全 因为已经加了锁
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);
break;
}
}
}
//针对红黑树的处理
else if (f instanceof TreeBin) {
Node<K,V> p;
binCount = 2;
//
//也就是存在key相同的并且允许被覆盖就覆盖旧的值 不然就根据红黑树的规则添加到树中
//putTreeVal() 解决hash冲突的逻辑和链表一样 但是会涉及到树的左旋和右旋
if ((p = ((TreeBin<K,V>)f).putTreeVal(hash, key,
value)) != null) {
oldVal = p.val;
if (!onlyIfAbsent)
p.val = value;
}
}
else if (f instanceof ReservationNode)
throw new IllegalStateException("Recursive update");
}
}
if (binCount != 0) {
//查看链表长度是否达到了树化的阈值（默认是8）
// 注意 这里并不是说达到了阈值就会树化
//而是要满足数组长度大于64而且链表长度大于等于阈值两个条件才会树化 否则会先进行扩容来减少链表的长度
if (binCount >= TREEIFY_THRESHOLD)
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
addCount(1L, binCount);
return null;
}initTable()方法初始化数组，为了保证只有一个线程进行初始化操作，利用cas保证线程安全问题。
代码语言：javascript复制 private final Node<K,V>[] initTable() {
Node<K,V>[] tab; int sc;
while ((tab = table) == null || tab.length == 0) {
if ((sc = sizeCtl) < 0)
//如果是小于0（有别的线程已经修改sizeCtl成功,所以sizeCtl为-1代表正在初始化数组）
//证明已经有线程在进行初始化 那么让出cpu时间片
Thread.yield(); // lost initialization race; just spin
//利用cas进行加锁 如果sizeCtl被修改为-1 修改成功，才能进行初始化 变相的一种加锁方式
else if (U.compareAndSetInt(this, SIZECTL, sc, -1)) {
try {
if ((tab = table) == null || tab.length == 0) {
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
@SuppressWarnings("unchecked")
//初始化数组 并且赋值给table成员变量
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
table = tab = nt;
//sc 赋值为数组扩容的阈值 就是 原容量 - 原容量/4;
sc = n - (n >>> 2);
}
} finally {
//这里扩容完成 sizeCtl也会变成数组扩容的阈值
sizeCtl = sc;
}
break;
}
}
return tab;
}接下里会分别对treeifyBin()方法和addcount()方法进行分析。
treeifyBin()方法
当链表长大于8的时候会调用treeifyBin方法，首先会对当前数组的长度进行判断，如果数组长度小于64，那么进行扩容的操作，如果大于64，那么将链表转化为红黑树。
所以链表转化为红黑树应该是要满足两个条件：
1.链表长度大于8
2.node数组长度大于6
代码语言：javascript复制private final void treeifyBin(Node<K,V>[] tab, int index) {
Node<K,V> b; int n;
if (tab != null) {
//如果数组长度小于64 那么进行扩容
if ((n = tab.length) < MIN_TREEIFY_CAPACITY)
//扩容 默认是扩大为原来容量的两倍
tryPresize(n << 1);
else if ((b = tabAt(tab, index)) != null && b.hash >= 0) {
//数组长度大于64 进行树化
//同样只是针对node加锁 严格控制锁的粒度
synchronized (b) {
//将链表转化为红黑树
if (tabAt(tab, index) == b) {
TreeNode<K,V> hd = null, tl = null;
for (Node<K,V> e = b; e != null; e = e.next) {
TreeNode<K,V> p =
new TreeNode<K,V>(e.hash, e.key, e.val,
null, null);
if ((p.prev = tl) == null)
hd = p;
else
tl.next = p;
tl = p;
}
//将数组下标的值设置为转化完的红黑树引用
setTabAt(tab, index, new TreeBin<K,V>(hd));
}
}
}
}
}数组长度小于64会调用tryPresize()方法进行扩容
扩容流程图
源码分析：首先计算扩容之后的容量，判断node数组是否进行初始化，没有初始化需要先进行初始化，然后会调用transfer()方法进行扩容，concurrentHashMap在扩容的时候设计的非常巧妙，当其他线性调用api的是否如果发现concurrentHashMap正在扩容，会调用helpTransfer方法协助数据的迁移，提高扩容的效率。
代码语言：javascript复制private final void tryPresize(int size) {
//判断扩容后的容量是否大于最大容量的一半 大于的话直接扩大为最大容量
//tableSizeFor 用来保证数组一定是2的n次幂
int c = (size >= (MAXIMUM_CAPACITY >>> 1)) ? MAXIMUM_CAPACITY :
tableSizeFor(size (size >>> 1) 1);
int sc;
while ((sc = sizeCtl) >= 0) {
Node<K,V>[] tab = table; int n;
//如果数组还没有初始化 先进行初始化
if (tab == null || (n = tab.length) == 0) {
n = (sc > c) ? sc : c;
//同样是利用cas的方式进行加锁 替换成功才会执行
//这个代码和initTable()的代码基本一样
if (U.compareAndSetInt(this, SIZECTL, sc, -1)) {
try {
if (table == tab) {
//创建新的数组 并赋值给table成员变量
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
table = nt;
sc = n - (n >>> 2);
}
} finally {
sizeCtl = sc;
}
}
}
//如果扩容之后的容量大于最大容量 直接返回
else if (c <= sc || n >= MAXIMUM_CAPACITY)
break;
else if (tab == table) {
//这里进行扩容操作 先计算扩容戳 这个扩容戳保证每次扩容都是唯一的
int rs = resizeStamp(n);
// 利用cas进行加锁
if (U.compareAndSetInt(this, SIZECTL, sc,
(rs << RESIZE_STAMP_SHIFT) 2))
//真正扩容的方法
transfer(tab, null);
}
}
}transfer()方法分析
transfer()方法是真正的扩容方法，会创建一个容量为原数组容量的2倍的新node数组，并且将原数组的元素迁移到新的数组中，并且把新的数组赋值给成员变量table。
concurrentHashMap支持多线程参与数据的迁移，首先计算每个线程负责的数据迁移的范围， 迁移的时候会把当前迁移的node修改为ForwardingNode，（其他线程判断当前节点如果是ForwardingNode节点，就跳过，避免重复迁移。）然后分别针对链表和红黑树两种情况进行数据的迁移。
1.链表：
遍历链表计算每个节点在新数组的下标位置，计算出的结果只有两种情况
a.计算出的新数组下标和原数组下标位置一样
b.计算出的新数组下标为(原数组下标 原数组的容量)
如果是a情况，将数据存在低位链表，如果是b情况存在在高位链表，然后赋值在新数组对应的下标位置。
2.红黑树：遍历红黑树，计算每个节点在新数组的下标位置，计算的结果也是一样的两种结果 如果是a情况，将数据存在低位树，如果是b情况存在在高位树，如果高位树或低位树的长度小于6，那么将高位树或低位树为链表，然后分别赋值到新数组中
代码语言：javascript复制private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
int n = tab.length, stride;
// 根据cpu数量计算一个线程迁移的范围 默认是16
//也就是说一个容量为32的旧数组 第一个线程第一次迁移的范围应该是16-31
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
stride = MIN_TRANSFER_STRIDE; // subdivide range
if (nextTab == null) { // initiating
//新数组为空 那么创建一个新的数组 容量是原来的两倍
try {
@SuppressWarnings("unchecked")
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表示数据迁移的开始索引 默认是从原数组的末尾开始迁移
//所以一个容量为32的旧数组 第一个线程第一次迁移的范围应该是16-31
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;
//前三个判断都是判断当前线程是否完成自己范围内要迁移的数据
//如果完成了直接设置 advance = false 跳出while循环
if (--i >= bound || finishing)
advance = false;
else if ((nextIndex = transferIndex) <= 0) {
i = -1;
advance = false;
}
else if (U.compareAndSetInt
(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) {
//如果已经扩容完成 将新数组赋值给table
nextTable = null;
table = nextTab;
sizeCtl = (n << 1) - (n >>> 1);
return;
}
//当前参与数据迁移的线程数-1 用cas保证数据的安全性
if (U.compareAndSetInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
//如果(sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT 条件成立 那么表示已经迁移完成
//因为在开始迁移数据的时候有对sizeCtl赋值 U.compareAndSetInt(this, SIZECTL, sc,(rs << RESIZE_STAMP_SHIFT) 2))
if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)
return;
finishing = advance = true;
i = n; // recheck before commit
}
}
else if ((f = tabAt(tab, i)) == null)
//如果是空的节点 直接cas替换为fwd节点 如果节点是fwd 也就是MOVE类型的节点 证明节点上有线程正在迁移
advance = casTabAt(tab, i, null, fwd);
else if ((fh = f.hash) == MOVED)
//如果是MOVE 那么就跳过 因为已经有线程在迁移了 不需要当前线程再去迁移
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);
//将旧数组的位置设置为fwd节点 表示该节点已经迁移数据
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;
}
}
//如果迁移后的高低位链表长度小于6（UNTREEIFY_THRESHOLD） 那么会把原来的数结构变成链表结构
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);
//将旧数组的位置设置为fwd节点 表示该节点已经迁移数据
setTabAt(tab, i, fwd);
advance = true;
}
}
}
}
}
}helpTransfer()
其他线程在对concurrentHashMap的操作的时候，如果发现concurrentHashMap正在扩容，那么会调用helpTransfer协助扩容
代码语言：javascript复制final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
Node<K,V>[] nextTab; int sc;
//进一步判断是否正在进行数据的迁移 如果是那么协助数据
if (tab != null && (f instanceof ForwardingNode) &&
(nextTab = ((ForwardingNode<K,V>)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;
//利用cas将负责迁移的线程数 1
if (U.compareAndSwapInt(this, SIZECTL, sc, sc 1)) {
// 调用transfer
transfer(tab, nextTab);
break;
}
}
return nextTab;
}
return table;
}addCount()方法
1.流程图
2.源码分析
在计算数组当前元素个数的时候，会使用CounterCell数组和baseCount进行计算，首先通过cas替换baseCount的值，如果失败，那么代表竞争很激烈，会利用CounterCell数组进行计算，每个线程使用CounterCell数组中一个元素进行累加个数，最后总的个数是CounterCell数组中每个元素的和加上baseCount。
代码语言：javascript复制private final void addCount(long x, int check) {
CounterCell[] cs; long b, s;
//这里采用CounterCell数组的方式来缓解cas计算count的压力
//先对baseCount进行cas操作 如果成功 那么直接返回 失败的话 就走下面的逻辑
if ((cs = counterCells) != null ||
!U.compareAndSetLong(this, BASECOUNT, b = baseCount, s = b x)) {
CounterCell c; long v; int m;
boolean uncontended = true;
//利用ThreadLocalRandom.getProbe() 随机生成CounterCell数组的下标位置
//数组没有初始化 或者数组的值为null 或者cas 计算节点数 如果失败 调用fullAddCount()初始化并计算值
if (cs == null || (m = cs.length - 1) < 0 ||
(c = cs[ThreadLocalRandom.getProbe() & m]) == null ||
!(uncontended =
U.compareAndSetLong(c, CELLVALUE, v = c.value, v x))) {
fullAddCount(x, uncontended);
return;
}
if (check <= 1)
return;
s = sumCount();
}
// 这一段是判断加入元素之后是否需要扩容
if (check >= 0) {
Node<K,V>[] tab, nt; int n, sc;
//数组初始化完之后 sizeCtl代表的是扩容阈值 s >= sizeCtl 证明需要扩容
//这下面一段代码和扩容的代码一模一样
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.compareAndSetInt(this, SIZECTL, sc, sc 1))
transfer(tab, nt);
}
else if (U.compareAndSetInt(this, SIZECTL, sc,
(rs << RESIZE_STAMP_SHIFT) 2))
transfer(tab, null);
//计算数组当前节点数量
s = sumCount();
}
}
}fullAddCount()
首先cas初始化CounterCell数组，利用ThreadLocalRandom.getProbe()生成的随机数和CounterCell数组的长度计算下标，利用cas进行计算count，如果失败会将CounterCell数组扩容，然后再次计算。
代码语言：javascript复制private final void fullAddCount(long x, boolean wasUncontended) {
int h;
if ((h = ThreadLocalRandom.getProbe()) == 0) {
ThreadLocalRandom.localInit(); // force initialization
h = ThreadLocalRandom.getProbe();
wasUncontended = true;
}
boolean collide = false; // True if last slot nonempty
for (;;) {
CounterCell[] cs; CounterCell c; int n; long v;
//第一次进入方法counterCells肯定是 null 所以不会走这个if的逻辑
if ((cs = counterCells) != null && (n = cs.length) > 0) {
//当前线程所用CounterCell数组下标的值为null
if ((c = cs[(n - 1) & h]) == null) {
if (cellsBusy == 0) { // Try to attach new Cell
//进行初始化
CounterCell r = new CounterCell(x); // Optimistic create
//cas加锁进行操作
if (cellsBusy == 0 &&
U.compareAndSetInt(this, CELLSBUSY, 0, 1)) {
boolean created = false;
try { // Recheck under lock
CounterCell[] rs; int m, j;
if ((rs = counterCells) != null &&
(m = rs.length) > 0 &&
rs[j = (m - 1) & h] == null) {
rs[j] = r;
created = true;
}
} finally {
cellsBusy = 0;
}
if (created)
break;
continue; // Slot is now non-empty
}
}
collide = false;
}
else if (!wasUncontended) // CAS already known to fail
wasUncontended = true; // Continue after rehash
// cas计算count 计算成功直接跳出循环
else if (U.compareAndSetLong(c, CELLVALUE, v = c.value, v x))
break;
//
else if (counterCells != cs || n >= NCPU)
collide = false; // At max size or stale
else if (!collide)
collide = true;
//前面cas计算count没有成功 说明竞争很激烈 需要对counterCells数组进行扩容
//cellsBusy 字段就是用来cas加锁的一个标志 cas进行加锁操作扩容
else if (cellsBusy == 0 &&
U.compareAndSetInt(this, CELLSBUSY, 0, 1)) {
try {
// 对counterCells进行扩容 扩容为原来的2倍
if (counterCells == cs) // Expand table unless stale
counterCells = Arrays.copyOf(cs, n << 1);
} finally {
cellsBusy = 0;
}
collide = false;
//扩容成功 进入下一次循环
continue; // Retry with expanded table
}
h = ThreadLocalRandom.advanceProbe(h);
}
//首次进入方法 肯定是0 而且counterCells没有初始化 应该是null
// 以cas的方式进行加锁 保证只有一个线程来进行初始化
else if (cellsBusy == 0 && counterCells == cs &&
U.compareAndSetInt(this, CELLSBUSY, 0, 1)) {
boolean init = false;
try { // Initialize table
if (counterCells == cs) {
//默认创建一个大小为2的数组 并且为当前线程
CounterCell[] rs = new CounterCell[2];
//初始化当前线程需要的数组下标的CounterCell
rs[h & 1] = new CounterCell(x);
counterCells = rs;
init = true;
}
} finally {
cellsBusy = 0;
}
if (init)
break;
}
//最后如果还是没有计算成功 那么cas加到baseCount中
else if (U.compareAndSetLong(this, BASECOUNT, v = baseCount, v x))
break; // Fall back on using base
}
}sumCount()
计算size，CounterCell数组每个元素的和 baseCount就是当前concurrentHashMap的size
代码语言：javascript复制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;
}四：结束语
在阅读比较长的方法的时候建议分段进行查看,比如transfer()方法和fullAddCount()方法，有很多的if else 我们如果分段阅读，会比较好理解。
