jdk currentHashMap的实现与hashMap比较像，不过有Hji'dian几点的不同
1 concurrentHashMap是线程安全的，它提供了锁的机制，并且它的锁不想hashTable那样锁全表而是将数据分成多个段，只在相应的段进行加锁（head链表或者红黑树的头）因此具有更高的性能。
2 hashTable SynchronizedMap是对整体数据进行加锁的。
接下来重点说说的它的实现方式。
实现的方式跟hashMap很像都是通过数组+链表/红黑树（链表数据个数>8） 实现的所不同的是它实现了并发做的部分。
多线程主要实现的地方在初始化，以及多线程辅助扩容这块。
currentHashMap的初始化
private final Node<K, V>[] initTable() {
Node<K, V>[] tab;
int sc;
while ((tab = table) == null || tab.length == 0) {
if ((sc = sizeCtl) < 0)
Thread.yield(); // lost initialization race; just spin
else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) { //将SIZECTL变成-1告诉别的线程，我在干活你们先等等
try {
if ((tab = table) == null || tab.length == 0) { //再次判断是不是为null，否则会有将之前的数据抹掉的风险
int n = (sc > 0) ? sc : DEFAULT_CAPACITY;
@SuppressWarnings("unchecked")
Node<K, V>[] nt = (Node<K, V>[]) new Node<?, ?>[n];
table = tab = nt;
sc = n - (n >>> 2);
}
} finally {
sizeCtl = sc;
}
break;
}
}
return tab;
}
关键点在sizeCtl这个参数
/**
- Table initialization and resizing control. When negative, the
- table is being initialized or resized: -1 for initialization,
- else -(1 + the number of active resizing threads). Otherwise,
- when table is null, holds the initial table size to use upon
- creation, or 0 for default. After initialization, holds the
- next element count value upon which to resize the table.
*/
获取和释放锁的过程：
sizeCtl是一个用于同步多个线程的共享变量，如果它的当前值为负数，则说明table正在被某个线程初始化或者扩容，所以，如果某个线程想要初始化table或者对table扩容，需要去竞争sizeCtl这个共享变量，获得变量的线程才有许可去进行接下来的操作，没能获得的线程将会一直自旋来尝试获得这个共享变量，所以获得sizeCtl这个变量的线程在完成工作之后需要设置回来，使得其他的线程可以走出自旋进行接下来的操作。而在initTable方法中我们可以看到，当线程发现sizeCtl小于0的时候，他就会让出CPU时间，而稍后再进行尝试，当发现sizeCtl不再小于0的时候，就会通过调用方法compareAndSwapInt来将sizeCtl共享变量变为-1，以告诉其他试图获得sizeCtl变量的线程，目前正在由本线程在享用该变量，在我完成我的任务之前你可以先休息一会，等会再来试试吧，我完成工作之后会释放掉的。
初始化完成后的再次校验：
在完成初始化table的任务之后，线程需要将sizeCtl设置成可以使得其他线程获得变量的状态，这其中还有一个地方需要注意，就是在某个线程通过U.compareAndSwapInt方法设置了sizeCtl之前和之后进行了两次check，来检测table是否被初始化过了，这种检测是必须的，因为在并发环境下，可能前一个线程正在初始化table但是还没有成功初始化，也就是table依然还为null，而有一个线程发现table为null他就会进行竞争sizeCtl以进行table初始化，但是当前线程在完成初始化之后，那个试图初始化table的线程获得了sizeCtl，但是此时table已经被初始化了，所以，如果没有再次判断的话，可能会将之后进行put操作的线程的更新覆盖掉，这是极为不安全的行为。
putVal的过程
final V putVal(K key, V value, boolean onlyIfAbsent) {
if (key == null || value == null) throw new NullPointerException();
int hash = spread(key.hashCode());//此处进行了一次重哈希（高16位与低16位异或），减少哈希碰撞
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();
else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) { // 如果插入的位置还没有节点，则使用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) // 如果要插入的位置的节点是转移节点，说明Map正在扩容，则协助扩容
tab = helpTransfer(tab, f);
else {
V oldVal = null;
synchronized (f) { // 对节点f加锁
if (tabAt(tab, i) == f) { // 加锁成功
if (fh >= 0) { // hashCode>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)))) { //key在Map中已经存在，更新key的值
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)//如果链表节点数量大于阈值（默认8），则转化为红黑树
treeifyBin(tab, i);
if (oldVal != null)
return oldVal;
break;
}
}
}
addCount(1L, binCount); // 节点插入完成后，更新Map中节点总数
return null;
}
1 首先是根据key的hashCode做一次重哈希（进一步减少哈希碰撞）先判断table为空，空则初始化Map，否则：
根据hashCode取模定位到table中的某个节点f，如果f为空，则新创建一个节点，使用cas方式更新到数组的f节点上，插入结束，
2 若f是转移转移节点，则调用helpTransfer协助转移，
3 锁定节点f（通过synchronized加锁）
4 节点f锁定成功后判断节点f类型，如果f是链表节点，则直接插入到链表底端（key不存在的话），如果节点f是红黑树节点，则按照二叉搜索树的方式插入节点，并调整树结构使其满足红黑规则
tabAt :获取index为i的 节点
casTabAt:table中index为i的节点更新为v（cas原子更新方式）
helpTransfer的最终的实现是交由transfer方法实现的，我们来看看是怎么做的：
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
int n = tab.length, stride;
//根据table的长度及cpu核数计算转移任务步长
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE) // 计算转移步长并判断是否小于最小转移步长
stride = MIN_TRANSFER_STRIDE; // subdivide range
if (nextTab == null) { // 第一个扩容线程进来需要初始化nextTable
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 = 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;
}
//通过cas方式尝试获取一个转移任务（transferIndex - 转移步长stride），获取成功后得到处理的下界及当前索引
else if (U.compareAndSwapInt
(this, TRANSFERINDEX, nextIndex,
nextBound = (nextIndex > stride ?
nextIndex - stride : 0))) {
bound = nextBound; // 更新当前子任务的下界
i = nextIndex - 1; // 更新当前index位置
advance = false;
}
}
if (i < 0 || i >= n || i + n >= nextn) { // 扩容结束
int sc;
if (finishing) { // 最后一个出去的线程：更新table指针及sizeCtl值
nextTable = null;
table = nextTab; // 指向扩容后的数组
sizeCtl = (n << 1) - (n >>> 1); //sizeCtl更新为最新的扩容阈值（2n - 0.5n = 1.5n = 2n * 0.75），移位实现保证高效率
return;
}
// sizeCtl减1，表示减少一个扩容线程
if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {
// 判断是否是最后一个扩容线程，如果不是则直接退出，由于第一个线程进来时把扩容戳rs左移16位+2更新到sizeCtl，所以如果是最后一个线程的话，sizeCtl -2 应该等于rs左移16位
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) { // 节点HasCode大于0 代表该节点为链表节点
// 由于数组长度n为2的幂次方，所以当数组长度增加到2n时，原来hash到table中i的数据节点在长度为2n的table中要么在低位nextTab[i]处，要么在高位nextTab[n+i]处，具体在哪个位置与(fh & n)的计算结果有关
int runBit = fh & n;
Node<K,V> lastRun = f;
// 此处循环的目的是找到链表中最后一个从低索引位置变到高索引位置或者从高索引位置变到低索引位置的节点lastRun，从lastRun节点到链表的尾节点可根据runBit直接插入到新数组nextTable的节点中，其目的是尽量减少新创建节点数量，直接更新指针位置
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;
}
// 对于lastRun之前的链表节点，根据hashCode&n可确定即将转移到nextTable中的低索引位置节点（nextTab[i]）还是高索引位置节点（nextTab[i + n]），并形成两个新的链表
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);
}
// 使用cas方式更新两个链表到新数组nextTable中，并且把原来的table节点i中的数值变为转移节点
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;
}
}
}
}
}
}
1 第一个扩容线程进来后创建nextTable数组，并设置transferIndex；
2 线程（第一个或其他）通过transferIndex-stride（扩容步长）来领取一个扩容子任务，transferIndex减到0说明所有子任务领取完成；
3 线程领取到扩容子任务后设置当前处理子任务的下界并更新当前处理节点所在的索引位置；
4 对子任务中的每个节点，扩容线程从后向前依次判断该节点是否已经转移，如果没有转移，则对该节点进行加锁，并且把节点对应的链表或红黑树转移到新数组nextTable中去；
5 如果线程处理的节点索引已经到达子任务的下界，则子任务执行结束，并尝试去领取新的子任务，若领取不到再判断当前线程是否是最后一个扩容线程，若是则最后扫描一遍数组，执行清理工作，否则直接退出。
本文由 妖言君 创作，采用 知识共享署名4.0 国际许可协议进行许可
本站文章除注明转载/出处外，均为本站原创或翻译，转载前请务必署名
最后编辑时间为: Jan 10, 2021 at 03:02 pm