首先将数据分为一段一段的存储，然后给每一段数据配一把锁，当一个线程占用锁访问其中一个段数据 时，其他段的数据也能被其他线程访问。
ConcurrentHashMap 是由 Segment 数组结构和 HashEntry 数组结构组成。
Segment 继承了 ReentrantLock,所以 Segment 是一种可重入锁，扮演锁的⻆色。HashEntry 用于存储 键值对数据。
一个 ConcurrentHashMap 里包含一个 Segment 数组。Segment 的结构和 HashMap 类似，是一种数组 和链表结构，一个 Segment 包含一个 HashEntry 数组，每个 HashEntry 是一个链表结构的元素，每 个 Segment 守护着一个 HashEntry 数组里的元素，当对 HashEntry 数组的数据进行修改时，必须首 先获得对应的 Segment 的锁。
对于ConcurrentHashMap的数据插入，这里要进行两次Hash去定位数据的存储位置
static class Segment<K,V> extends ReentrantLock implements Serializable {
}从上Segment的继承体系可以看出，Segment实现了ReentrantLock,也就带有锁的功能，当执行put操作时，会进行第一次key的hash来定位Segment的位置，如果该Segment还没有初始化，即通过CAS操作进行赋值，然后进行第二次hash操作，找到相应的HashEntry的位置，这里会利用继承过来的锁的特性，在将数据插入指定的HashEntry位置时（链表的尾端），会通过继承ReentrantLock的tryLock（）方法尝试去获取锁，如果获取成功就直接插入相应的位置，如果已经有线程获取该Segment的锁，那当前线程会以自旋的方式去继续的调用tryLock（）方法去获取锁，超过指定次数就挂起，等待唤醒。
-
首先对key进行第一次hash，通过hash值确定segment的位置
-
然后在segment内进行操作，获取锁
-
接着获取当前segment的HashEntry数组，然后对key进行第二次hash，通过hash值确定在HashEntry数组的索引位置。
-
然后对当前索引的HashEntry链进行遍历，如果有重复的key，则替换；如果没有重复的，则插入
-
关闭锁 可见，在整个put过程中，进行了2次hash操作，才最终确定key的位置。
ConcurrentHashMap的get操作跟HashMap类似，只是ConcurrentHashMap第一次需要经过一次hash定位到Segment的位置，然后再hash定位到指定的HashEntry，遍历该HashEntry下的链表进行对比，成功就返回，不成功就返回null。
第一种方案他会使用不加锁的模式去尝试多次计算ConcurrentHashMap的size，最多三次，比较前后两次计算的结果，结果一致就认为当前没有元素加入，计算的结果是准确的
第二种方案是如果第一种方案不符合，他就会给每个Segment加上锁，然后计算ConcurrentHashMap的size返回
ConcurrentHashMap不允许空值空键
ConcurrentHashMap 取消了 Segment 分段锁，采用 CAS 和 synchronized 来保证并发安全。数据结构 跟 HashMap1.8 的结构类似，数组+链表/红黑二叉树。Java 8 在链表⻓度超过一定阈值(8)时将链表 (寻址时间复杂度为 O(N))转换为红黑树(寻址时间复杂度为 O(log(N)))
synchronized 只锁定当前链表或红黑二叉树的首节点，这样只要 hash 不冲突，就不会产生并发，效 率又提升 N 倍。
private transient volatile int sizeCtl;sizeCtl在初始化时一直出现，在扩容的时候也有使用。
- sizeCtl为0，代表数组未初始化，且数组的初始容量为16
- sizeCtl为正数，如果数组未初始化，那么其记录的是数组的初始容量，如果数组已经初始化，那么其记录的是数组的扩容阈值。
- sizeCtl为-1，表示数组正在进行初始化
- sizeCtl小于0，并且不是-1，表示数组正在扩容，-(1 + n)表示此时有n个线程正在共同完成数组的扩容操作。
static final int HASH_BITS = 0x7fffffff; // usable bits of normal node hash
static final int spread(int h) {
return (h ^ (h >>> 16)) & HASH_BITS;
}spread(key.hashCode());跟hashmap类似，高16位和低16位做异或操作，增加随机性减少冲突，这里还把hash值限制在Integer.MAX_VALUE之内。
private static final sun.misc.Unsafe U;
// 直接操作内存
U = sun.misc.Unsafe.getUnsafe();
/**
* 寻找指定数组在内存中i位置的数据
*/
@SuppressWarnings("unchecked")
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);
}/**
* 将结点v插入tab[i]，采用cas的方法，c是期望值，如果当前值不等于期望值就一直循环尝试插入
* @param tab
* @param i
* @param c
* @param v
* @param <K>
* @param <V>
* @return
*/
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);
}
/**
* 将结点v设置到tab[i]的位置
* @param tab
* @param i
* @param v
* @param <K>
* @param <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);
}setTabAt都是在加锁的代码里面调用的，不然就使用casTabAt，无锁操作。
- 如果没有初始化就先调用initTable()方法来进行初始化过程
- 如果没有hash冲突就直接CAS插入
- 如果还在进行扩容操作就先进行协作扩容
- 如果存在hash冲突，就加锁来保证线程安全，这里有两种情况，一种是链表形式就直接遍历到尾端插入，一种是红黑树就按照红黑树结构插入，
- 最后一个如果Hash冲突时会形成Node链表，在链表长度超过8，Node数组超过64时会将链表结构转换为红黑树的结构，break退出循环
- 如果添加成功就调用addCount()方法统计size，并且检查是否需要扩容
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());
// 统计桶中数量
int binCount = 0;
for (Node<K,V>[] tab = table;;) {
Node<K,V> f; int n, i, fh;
// tab=null，就初始化table
if (tab == null || (n = tab.length) == 0)
tab = initTable();
// tab[i]=null,就cas插入key、value
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
}
// 如果正在扩容，就当前线程就去协助扩容
else if ((fh = f.hash) == MOVED)
tab = helpTransfer(tab, f);
// 存在hash冲突
else {
V oldVal = null;
synchronized (f) {
if (tabAt(tab, i) == f) {
if (fh >= 0) {
binCount = 1;
for (Node<K,V> e = f;; ++binCount) {
K ek;
// 更新value值
if (e.hash == hash &&
((ek = e.key) == key ||
(ek != null && key.equals(ek)))) {
oldVal = e.val;
if (!onlyIfAbsent)
e.val = value;
break;
}
// 不存在这个key，就插入一个新结点，尾插法
// pred是e的前驱结点
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;
}
}
}
// 结点数加1
addCount(1L, binCount);
return null;
}初始化数组，只允许一个数组进行初始化，实现原理，利用sizeCtl，sizeCtl默认为0，
- while循环判断数组是否为null，且数组长度是否为0
- 在while内部，先判断sizeCtl是否小于0，如果小于0就让出cpu，说明别的线程正在初始化
- 如果sizeCtl不小于0，就进行cas操作，就sizeCtl设置为-1，设置成功就再判断数组是否为null，且数组长度是否为0，如果满足条件，就进行初始化，并且把sizeCtl设置为n/2；如果不满足就退出循环
- 初始化完数组，别的线程如果还在initTable函数里，就会尝试把sizeCtl设置为-1，设置成功，判断数组不为null，退出循环。
/**
* Initializes table, using the size recorded in sizeCtl.
* 初始化数组
*/
private final Node<K,V>[] initTable() {
Node<K,V>[] tab; int sc;
// 如果当前table数组是空的
while ((tab = table) == null || tab.length == 0) {
// sizeCtl默认为0，如果sizeCtl<0，表示有线程正在初始化，当前线程让出cpu，
// 而且此时sizeCtl=-1，因为此时在initTable函数里，看下面代码
if ((sc = sizeCtl) < 0)
Thread.yield(); // lost initialization race; just spin
// compareAndSwapInt，cas操作，当前sizeCtl如果等于期望值sc,就将sizeCtl设置为-1
// 此时就可以看到，只能有一个线程进行初始化，别的线程再调用initTable，就会进入第一个
// if让出cpu
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) {
try {
// 这里再判断table是否为null,相当于DCL双重检查
if ((tab = table) == null || tab.length == 0) {
// 如果sc=0，就会使用默认容量
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n];
table = tab = nt;
// sc=n/2
sc = n - (n >>> 2);
}
} finally {
// 此时sizeCtl记录的是扩容阈值
sizeCtl = sc;
}
break;
}
}
return tab;
}ConcurrentHashMap是多线程动态扩容，一个线程负责一段数组，将老数组里面的元素添加到新数组里面。
/**
* Helps transfer if a resize is in progress.
* 协助扩容函数
*/
final Node<K,V>[] helpTransfer(Node<K,V>[] tab, Node<K,V> f) {
Node<K,V>[] nextTab; int sc;
// 如果 table 不是空 且 node 节点是转移类型，数据检验
// 且 node 节点的 nextTable（新 table） 不是空，同样也是数据校验
// 尝试帮助扩容
if (tab != null && (f instanceof ForwardingNode) &&
(nextTab = ((ForwardingNode<K,V>)f).nextTable) != null) {
// 这个方法的返回一个与table容量n大小有关的扩容标记
int rs = resizeStamp(tab.length);
// 如果 nextTab 没有被并发修改 且 tab 也没有被并发修改
// 且 sizeCtl < 0 （说明还在扩容）
while (nextTab == nextTable && table == tab &&
(sc = sizeCtl) < 0) {
// 如果 sizeCtl 无符号右移 16 不等于 rs （ sc前 16 位如果不等于标识符，则标识符变化了）
// 或者 sizeCtl == rs + 1 （扩容结束了，不再有线程进行扩容）（默认第一个线程设置 sc ==rs 左移 16 位 + 2，当第一个线程结束扩容了，就会将 sc 减一。这个时候，sc 就等于 rs + 1）
// 或者 sizeCtl == rs + 65535 （如果达到最大帮助线程的数量，即 65535）
// 或者转移下标正在调整 （扩容结束）
// 结束循环，返回 table
if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 ||
sc == rs + MAX_RESIZERS || transferIndex <= 0)
break;
// 如果以上都不是, 将 sizeCtl + 1, （表示增加了一个线程帮助其扩容）
if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) {
// 进行转移
transfer(tab, nextTab);
break;
}
}
return nextTab;
}
return table;
} /**
* Moves and/or copies the nodes in each bin to new table. See
* above for explanation.
* 复制结点到一个新数组
*/
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
int n = tab.length, stride;
// stride表示每个线程处理的槽个数。同时基于性能考虑，stride不能比16小。
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
stride = MIN_TRANSFER_STRIDE; // subdivide range
if (nextTab == null) { // initiating
try {
// 如果nextTab=null就新建一个table
@SuppressWarnings("unchecked")
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
nextTab = nt;
} catch (Throwable ex) { // try to cope with OOME
// 数组扩容的时候有可能出现OOME，这时需要将sizeCtl设置为Integer.MAX_VALUE，用以表示这个异常
sizeCtl = Integer.MAX_VALUE;
return;
}
nextTable = nextTab;
// transferIndex用以标识扩容总的进度，默认设置为原数组长度。
// 因为扩容时候的元素迁移是从最末的元素开始的，所以迁移的时候下标是递减的，从下面的`--i`就能看出来了
transferIndex = n;
}
int nextn = nextTab.length;
// 扩容时准备的特殊节点
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
// 用于表示table里面单个元素是否迁移完毕，初始为true是为了进入while循环。
// 1. true表示还未迁移完
// 2. false表示已经迁移完
boolean advance = true;
// 用于表示table里面的所有元素是否迁移完毕
boolean finishing = false; // to ensure sweep before committing nextTab
// 这儿又是自旋，bound表示槽的边界
for (int i = 0, bound = 0;;) {
Node<K,V> f; int fh;
while (advance) {
int nextIndex, nextBound;
// 倒序迁移旧元素的下标已达到槽的边界，或者整个table已经迁移完毕，说明迁移完成了
if (--i >= bound || finishing)
advance = false;
// 扩容总进度<=0，说明迁移完成了
else if ((nextIndex = transferIndex) <= 0) {
i = -1;
advance = false;
}
// transferIndex减去每个线程处理的槽个数，将这个值设置为即将迁移的边界
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=扩容阈值
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
}
}
// 数组元素所在的位置没有值，则直接设置ForwordNode
else if ((f = tabAt(tab, i)) == null)
advance = casTabAt(tab, i, null, fwd);
// 数组元素所在位置节点为ForwordNode，表示已经被迁移过了
else if ((fh = f.hash) == MOVED)
advance = true; // already processed
else {
// 其他情况，需要对元素首结点进行加锁，然后将该元素所在槽里面的每个结点都进行迁移
synchronized (f) {
if (tabAt(tab, i) == f) {
// ln表示扩容后索引位置不变的情况，hn表示扩容后索引位置要变成当前位置+原容量的情况
Node<K,V> ln, hn;
if (fh >= 0) {
// runBit = 0表示扩容后索引位置不变，runBit=1表示扩容后索引位置要变成当前位置+原容量的情况
int runBit = fh & n;
// 用lastRun记录下最后一个需要遍历的结点，注意，lastRun不一定是最后一个结点，lastRun后面的结点
// 跟lastRun扩容后的位置是一样的，所以直接把这个小链表插入到新数组就行，不用再挨个遍历后面的结点
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是否相同，如果相同，就不用记录了，因为之前可能已经记录过lastRun了
// 如果不同，则更改lastRun。lastRun结点到尾结点这段结点的扩容后索引位置相同
if (b != runBit) {
runBit = b;
lastRun = p;
}
}
// ln表示扩容后索引位置不变的情况
if (runBit == 0) {
ln = lastRun;
hn = null;
}
// hn表示扩容后索引位置要变成当前位置+原容量的情况
else {
hn = lastRun;
ln = null;
}
// 这个for循环只用遍历到lastRun结点，因为lastRun后面的结点跟lastRun结点插入位置相同
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);
}
// 将ln链表插入到nextTab[i]位置
setTabAt(nextTab, i, ln);
// 将hn链表插入到nextTab[i + n]位置
setTabAt(nextTab, i + n, hn);
// TODO 我觉得这句代码可能是处理ln=null情况，将fwd插入到tab[i]
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;
}
}
}
}
}
}- 计算hash值，定位到该table索引位置，如果是首节点符合就返回
- 如果遇到扩容的时候，会调用标志正在扩容节点ForwardingNode的find方法，查找该节点，匹配就返回
- 以上都不符合的话，就往下遍历节点，匹配就返回，否则最后就返回null
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) {
// 判断tab[i]的第一个结点是不是要找的结点
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;
}添加容量并没有加锁，而是使用cas方式，往一个数组里面的+1，最后将整个数组求和，就是当前ConcurrentHashMap的大小。
private transient volatile CounterCell[] counterCells;
@sun.misc.Contended static final class CounterCell {
volatile long value;
CounterCell(long x) { value = x; }
} 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;
}放在这里就太长了，我单独放在一个文件里面了。注释版的ConcurrentHashMap