标签：实现 mda 完成 length 并发 依次 before ble int()
ConcurrentHashMap 是并发中的重中之重，也是最常用的数据结果，之前的文章中，我们介绍了 putVal 方法。并发编程之 ConcurrentHashMap（JDK 1.8） putVal 源码分析。其中分析了 initTable 方法和 putVal 方法，但也留下了一句话：
这篇文章仅仅是 ConcurrentHashMap 的开头，关于 ConcurrentHashMap 里面的精华太多，值得我们好好学习。
说道精华，他的扩容方法绝对是精华，要知道，ConcurrentHashMap 扩容是高度并发的。
今天来逐行分析源码。
首先说结论。源码加注释我会放在后面。该方法的执行逻辑如下：
通过计算 CPU 核心数和 Map 数组的长度得到每个线程（CPU）要帮助处理多少个桶，并且这里每个线程处理都是平均的。默认每个线程处理 16 个桶。因此，如果长度是 16 的时候，扩容的时候只会有一个线程扩容。
初始化临时变量 nextTable。将其在原有基础上扩容两倍。
死循环开始转移。多线程并发转移就是在这个死循环中，根据一个 finishing 变量来判断，该变量为 true 表示扩容结束，否则继续扩容。
3.1 进入一个 while 循环，分配数组中一个桶的区间给线程，默认是 16. 从大到小进行分配。当拿到分配值后，进行 i-- 递减。这个 i 就是数组下标。（其中有一个 bound 参数，这个参数指的是该线程此次可以处理的区间的最小下标，超过这个下标，就需要重新领取区间或者结束扩容，还有一个 advance 参数，该参数指的是是否继续递减转移下一个桶，如果为 true，表示可以继续向后推进，反之，说明还没有处理好当前桶，不能推进)
3.2 出 while 循环，进 if 判断，判断扩容是否结束，如果扩容结束，清空临死变量，更新 table 变量，更新库容阈值。如果没完成，但已经无法领取区间（没了），该线程退出该方法，并将 sizeCtl 减一，表示扩容的线程少一个了。如果减完这个数以后，sizeCtl 回归了初始状态，表示没有线程再扩容了，该方法所有的线程扩容结束了。（这里主要是判断扩容任务是否结束，如果结束了就让线程退出该方法，并更新相关变量）。然后检查所有的桶，防止遗漏。
3.3 如果没有完成任务，且 i 对应的槽位是空，尝试 CAS 插入占位符，让 putVal 方法的线程感知。
3.4 如果 i 对应的槽位不是空，且有了占位符，那么该线程跳过这个槽位，处理下一个槽位。
3.5 如果以上都是不是，说明这个槽位有一个实际的值。开始同步处理这个桶。
3.6 到这里，都还没有对桶内数据进行转移，只是计算了下标和处理区间，然后一些完成状态判断。同时，如果对应下标内没有数据或已经被占位了，就跳过了。
处理每个桶的行为都是同步的。防止 putVal 的时候向链表插入数据。
4.1 如果这个桶是链表，那么就将这个链表根据 length 取于拆成两份，取于结果是 0 的放在新表的低位，取于结果是 1 放在新表的高位。
4.2 如果这个桶是红黑数，那么也拆成 2 份，方式和链表的方式一样，然后，判断拆分过的树的节点数量，如果数量小于等于 6，改造成链表。反之，继续使用红黑树结构。
4.3 到这里，就完成了一个桶从旧表转移到新表的过程。
好，以上，就是 transfer 方法的总体逻辑。还是挺复杂的。再进行精简，分成 3 步骤：
大体就是上的的 3 个步骤。
再来看看源码和注释。
源码加注释：
/**
* Moves and/or copies the nodes in each bin to new table. See
* above for explanation.
*
* transferIndex 表示转移时的下标，初始为扩容前的 length。
*
* 我们假设长度是 32
*/
private final void transfer(Node<K,V>[] tab, Node<K,V>[] nextTab) {
int n = tab.length, stride;
// 将 length / 8 然后除以 CPU核心数。如果得到的结果小于 16，那么就使用 16。
// 这里的目的是让每个 CPU 处理的桶一样多，避免出现转移任务不均匀的现象，如果桶较少的话，默认一个 CPU（一个线程）处理 16 个桶
if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE)
stride = MIN_TRANSFER_STRIDE; // subdivide range 细分范围 stridea：TODO
// 新的 table 尚未初始化
if (nextTab == null) { // initiating
try {
// 扩容 2 倍
Node<K,V>[] nt = (Node<K,V>[])new Node<?,?>[n << 1];
// 更新
nextTab = nt;
} catch (Throwable ex) { // try to cope with OOME
// 扩容失败， sizeCtl 使用 int 最大值。
sizeCtl = Integer.MAX_VALUE;
return;// 结束
}
// 更新成员变量
nextTable = nextTab;
// 更新转移下标，就是 老的 tab 的 length
transferIndex = n;
}
// 新 tab 的 length
int nextn = nextTab.length;
// 创建一个 fwd 节点，用于占位。当别的线程发现这个槽位中是 fwd 类型的节点，则跳过这个节点。
ForwardingNode<K,V> fwd = new ForwardingNode<K,V>(nextTab);
// 首次推进为 true，如果等于 true，说明需要再次推进一个下标（i--），反之，如果是 false，那么就不能推进下标，需要将当前的下标处理完毕才能继续推进
boolean advance = true;
// 完成状态，如果是 true，就结束此方法。
boolean finishing = false; // to ensure sweep before committing nextTab
// 死循环,i 表示下标，bound 表示当前线程可以处理的当前桶区间最小下标
for (int i = 0, bound = 0;;) {
Node<K,V> f; int fh;
// 如果当前线程可以向后推进；这个循环就是控制 i 递减。同时，每个线程都会进入这里取得自己需要转移的桶的区间
while (advance) {
int nextIndex, nextBound;
// 对 i 减一，判断是否大于等于 bound （正常情况下，如果大于 bound 不成立，说明该线程上次领取的任务已经完成了。那么，需要在下面继续领取任务）
// 如果对 i 减一大于等于 bound（还需要继续做任务），或者完成了，修改推进状态为 false，不能推进了。任务成功后修改推进状态为 true。
// 通常，第一次进入循环，i-- 这个判断会无法通过，从而走下面的 nextIndex 赋值操作（获取最新的转移下标）。其余情况都是：如果可以推进，将 i 减一，然后修改成不可推进。如果 i 对应的桶处理成功了，改成可以推进。
if (--i >= bound || finishing)
advance = false;// 这里设置 false，是为了防止在没有成功处理一个桶的情况下却进行了推进
// 这里的目的是：1. 当一个线程进入时，会选取最新的转移下标。2. 当一个线程处理完自己的区间时，如果还有剩余区间的没有别的线程处理。再次获取区间。
else if ((nextIndex = transferIndex) <= 0) {
// 如果小于等于0，说明没有区间了 ，i 改成 -1，推进状态变成 false，不再推进，表示，扩容结束了，当前线程可以退出了
// 这个 -1 会在下面的 if 块里判断，从而进入完成状态判断
i = -1;
advance = false;// 这里设置 false，是为了防止在没有成功处理一个桶的情况下却进行了推进
}// CAS 修改 transferIndex，即 length - 区间值，留下剩余的区间值供后面的线程使用
else if (U.compareAndSwapInt
(this, TRANSFERINDEX, nextIndex,
nextBound = (nextIndex > stride ?
nextIndex - stride : 0))) {
bound = nextBound;// 这个值就是当前线程可以处理的最小当前区间最小下标
i = nextIndex - 1; // 初次对i 赋值，这个就是当前线程可以处理的当前区间的最大下标
advance = false; // 这里设置 false，是为了防止在没有成功处理一个桶的情况下却进行了推进，这样对导致漏掉某个桶。下面的 if (tabAt(tab, i) == f) 判断会出现这样的情况。
}
}// 如果 i 小于0 （不在 tab 下标内，按照上面的判断，领取最后一段区间的线程扩容结束）
// 如果 i >= tab.length(不知道为什么这么判断)
// 如果 i + tab.length >= nextTable.length （不知道为什么这么判断）
if (i < 0 || i >= n || i + n >= nextn) {
int sc;
if (finishing) { // 如果完成了扩容
nextTable = null;// 删除成员变量
table = nextTab;// 更新 table
sizeCtl = (n << 1) - (n >>> 1); // 更新阈值
return;// 结束方法。
}// 如果没完成
if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) {// 尝试将 sc -1. 表示这个线程结束帮助扩容了，将 sc 的低 16 位减一。
if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT)// 如果 sc - 2 不等于标识符左移 16 位。如果他们相等了，说明没有线程在帮助他们扩容了。也就是说，扩容结束了。
return;// 不相等，说明没结束，当前线程结束方法。
finishing = advance = true;// 如果相等，扩容结束了，更新 finising 变量
i = n; // 再次循环检查一下整张表
}
}
else if ((f = tabAt(tab, i)) == null) // 获取老 tab i 下标位置的变量，如果是 null，就使用 fwd 占位。
advance = casTabAt(tab, i, null, fwd);// 如果成功写入 fwd 占位，再次推进一个下标
else if ((fh = f.hash) == MOVED)// 如果不是 null 且 hash 值是 MOVED。
advance = true; // already processed // 说明别的线程已经处理过了，再次推进一个下标
else {// 到这里，说明这个位置有实际值了，且不是占位符。对这个节点上锁。为什么上锁，防止 putVal 的时候向链表插入数据
synchronized (f) {
// 判断 i 下标处的桶节点是否和 f 相同
if (tabAt(tab, i) == f) {
Node<K,V> ln, hn;// low, height 高位桶，低位桶
// 如果 f 的 hash 值大于 0 。TreeBin 的 hash 是 -2
if (fh >= 0) {
// 对老长度进行与运算（第一个操作数的的第n位于第二个操作数的第n位如果都是1，那么结果的第n为也为1，否则为0）
// 由于 Map 的长度都是 2 的次方（000001000 这类的数字），那么取于 length 只有 2 种结果，一种是 0，一种是1
// 如果是结果是0 ，Doug Lea 将其放在低位，反之放在高位，目的是将链表重新 hash，放到对应的位置上，让新的取于算法能够击中他。
int runBit = fh & n;
Node<K,V> lastRun = f; // 尾节点，且和头节点的 hash 值取于不相等
// 遍历这个桶
for (Node<K,V> p = f.next; p != null; p = p.next) {
// 取于桶中每个节点的 hash 值
int b = p.hash & n;
// 如果节点的 hash 值和首节点的 hash 值取于结果不同
if (b != runBit) {
runBit = b; // 更新 runBit，用于下面判断 lastRun 该赋值给 ln 还是 hn。
lastRun = p; // 这个 lastRun 保证后面的节点与自己的取于值相同，避免后面没有必要的循环
}
}
if (runBit == 0) {// 如果最后更新的 runBit 是 0 ，设置低位节点
ln = lastRun;
hn = null;
}
else {
hn = lastRun; // 如果最后更新的 runBit 是 1， 设置高位节点
ln = null;
}// 再次循环，生成两个链表，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;
// 如果与运算结果是 0，那么就还在低位
if ((ph & n) == 0) // 如果是0 ，那么创建低位节点
ln = new Node<K,V>(ph, pk, pv, ln);
else // 1 则创建高位
hn = new Node<K,V>(ph, pk, pv, hn);
}
// 其实这里类似 hashMap
// 设置低位链表放在新链表的 i
setTabAt(nextTab, i, ln);
// 设置高位链表，在原有长度上加 n
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);
// 和链表相同的判断，与运算 == 0 的放在低位
if ((h & n) == 0) {
if ((p.prev = loTail) == null)
lo = p;
else
loTail.next = p;
loTail = p;
++lc;
} // 不是 0 的放在高位
else {
if ((p.prev = hiTail) == null)
hi = p;
else
hiTail.next = p;
hiTail = p;
++hc;
}
}
// 如果树的节点数小于等于 6，那么转成链表，反之，创建一个新的树
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;
}
}
}
}
}
}
代码加注释比较长，有兴趣可以逐行对照，有 2 个判断楼主看不懂为什么这么判断，知道的同学可以提醒一下。
然后，说说精华的部分。
假设总长度是 64 ，每个线程可以分到 16 个桶，各自处理，不会互相影响。
扩容前的状态。
当对 4 号桶或者 10 号桶进行转移的时候，会将链表拆成两份，规则是根据节点的 hash 值取于 length，如果结果是 0，放在低位，否则放在高位。
因此，10 号桶的数据，黑色节点会放在新表的 10 号位置，白色节点会放在新桶的 26 号位置。
下图是循环处理桶中数据的逻辑：
处理完之后，新桶的数据是这样的：
transfer 方法可以说很牛逼，很精华，内部多线程扩容性能很高，
通过给每个线程分配桶区间，避免线程间的争用，通过为每个桶节点加锁，避免 putVal 方法导致数据不一致。同时，在扩容的时候，也会将链表拆成两份，这点和 HashMap 的 resize 方法类似。
而如果有新的线程想 put 数据时，也会帮助其扩容。鬼斧神工，令人赞叹。
在之前的文章提到ConcurrentHashMap是一个线程安全的，那么我么看一下ConcurrentHashMap如何进行操作的。
ConcurrentHashMap与HashTable区别？
HashTable
put()源代码
从代码可以看出来在所有put 的操作的时候 都需要用 synchronized 关键字进行同步。并且key 不能为空。
这样相当于每次进行put 的时候都会进行同步 当10个线程同步进行操作的时候，就会发现当第一个线程进去 其他线程必须等待第一个线程执行完成，才可以进行下去。性能特别差。
CurrentHashMap
分段锁技术：ConcurrentHashMap相比 HashTable而言解决的问题就是 的 它不是锁全部数据，而是锁一部分数据，这样多个线程访问的时候就不会出现竞争关系。不需要排队等待了。
从图中可以看出来ConcurrentHashMap的主干是个Segment数组。
这就是为什么ConcurrentHashMap支持允许多个修改同时并发进行，原因就是采用的Segment分段锁功能，每一个Segment 都想的于小的hash table并且都有自己锁，只要修改不再同一个段上就不会引起并发问题。
final Segment<K,V>[] segments;
1
使用ConConcurrentHashMap时候 有时候会遇到跨段的问题，跨段的时候【size()、 containsValue()】，可能需要锁定部分段或者全段，当操作结束之后，又回按照 顺序 进行 释放 每一段的锁。注意是按照顺序解锁的。，每个Segment又包含了多个HashEntry.
transient volatile HashEntry<K,V>[] table;
static final class HashEntry<K,V> {
final int hash;
final K key;
volatile V value;
volatile HashEntry<K,V> next;
//其他省略
}
1
2
3
4
5
6
7
8
9
需要注意的是 Segment 是一种可重入锁（继承ReentrantLock)
那么我简单说一下ReentrantLock 与synchronized有什么区别？
synchronized 是一个同步锁 synchronized （this）
同步锁 当一个线程A 访问 【资源】的代码同步块的时候，A线程就会持续持有当前锁的状态，如果其他线程B-E 也要访问【资源】的代码同步块的时候将会收到阻塞，因此需要排队等待A线程释放锁的状态。（如图情况1）但是注意的是，当一个线程B-E 只是不能方法 A线程 【资源】的代码同步块，仍然可以访问其他的非资源同步块。
ReentrantLock 可重入锁 通常两类：公平性、非公平性
公平性：根据线程请求锁的顺序依次获取锁，当一个线程A 访问 【资源】的期间，线程A 获取锁资源，此时内部存在一个计数器num+1，在访问期间，线程B、C请求 资源时，发现A 线程在持有当前资源，因此在后面生成节点排队（B 处于待唤醒状态），假如此时a线程再次请求资源时，不需要再次排队，可以直接再次获取当前资源 （内部计数器+1 num=2） ，当A线程释放所有锁的时候（num=0），此时会唤醒B线程进行获取锁的操作，其他C-E线程就同理。（情况2）
非公平性：当A线程已经释放所之后，准备唤醒线程B获取资源的时候，此时线程M 获取请求，此时会出现竞争，线程B 没有竞争过M线程，测试M获取的线程因此，M会有限获得资源，B继续睡眠。（情况2）
synchronized 是一个非公平性锁。 非公平性 会比公平性锁的效率要搞很多原因，不需要通知等待。
ReentrantLock 提供了 new Condition可以获得多个Condition对象,可以简单的实现比较复杂的线程同步的功能.通过await(),signal()以实现。
ReentrantLock 提供可以中断锁的一个方法lock.lockInterruptibly()方法。
Jdk 1.8 synchronized和 ReentrantLock 比较的话，官方比较建议用synchronized。
在了解Segment 机制之后我们继续看一下ConcurrentHashMap核心构造方法代码。
// 跟HashMap结构有点类似
Segment(float lf, int threshold, HashEntry<K,V>[] tab) {
this.loadFactor = lf;//负载因子
this.threshold = threshold;//阈值
this.table = tab;//主干数组即HashEntry数组
}
1
2
3
4
5
6
7
构造方法
public ConcurrentHashMap(int initialCapacity,
float loadFactor, int concurrencyLevel) {
if (!(loadFactor > 0) || initialCapacity < 0 || concurrencyLevel <= 0)
throw new IllegalArgumentException();
//MAX_SEGMENTS 为1<<16=65536，也就是最大并发数为65536
if (concurrencyLevel > MAX_SEGMENTS)
concurrencyLevel = MAX_SEGMENTS;
int sshift = 0;
//ssize 为segments数组长度，concurrentLevel计算得出
int ssize = 1;
while (ssize < concurrencyLevel) {
++sshift;
ssize <<= 1;
}
//segmentShift和segmentMask这两个变量在定位segment时会用到
this.segmentShift = 32 - sshift;
this.segmentMask = ssize - 1;
if (initialCapacity > MAXIMUM_CAPACITY)
initialCapacity = MAXIMUM_CAPACITY;
//计算cap的大小，即Segment中HashEntry的数组长度，cap也一定为2的n次方.
int c = initialCapacity / ssize;
if (c * ssize < initialCapacity)
++c;
int cap = MIN_SEGMENT_TABLE_CAPACITY;
while (cap < c)
cap <<= 1;
//创建segments数组并初始化第一个Segment，其余的Segment延迟初始化
Segment<K,V> s0 =
new Segment<K,V>(loadFactor, (int)(cap * loadFactor),
(HashEntry<K,V>[])new HashEntry[cap]);
Segment<K,V>[] ss = (Segment<K,V>[])new Segment[ssize];
UNSAFE.putOrderedObject(ss, SBASE, s0);
this.segments = ss;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
从以上代码可以看出ConcurrentHashMap有比较重要的三个参数：
loadFactor 负载因子 0.75
threshold 初始 容量 16
concurrencyLevel 实际上是Segment的实际数量。
ConcurrentHashMap如何发生ReHash？
ConcurrentLevel 一旦设定的话，就不会改变。ConcurrentHashMap当元素个数大于临界值的时候，就会发生扩容。但是ConcurrentHashMap与其他的HashMap不同的是，它不会对Segmengt 数量增大，只会增加Segmengt 后面的链表容量的大小。即对每个Segmengt 的元素进行的ReHash操作。
我们再看一下核心的ConcurrentHashMapput ()方法：
public V put(K key, V value) {
Segment<K,V> s;
//concurrentHashMap不允许key/value为空
if (value == null)
throw new NullPointerException();
//hash函数对key的hashCode重新散列，避免差劲的不合理的hashcode，保证散列均匀
int hash = hash(key);
//返回的hash值无符号右移segmentShift位与段掩码进行位运算，定位segment
int j = (hash >>> segmentShift) & segmentMask;
if ((s = (Segment<K,V>)UNSAFE.getObject // nonvolatile; recheck
(segments, (j << SSHIFT) + SBASE)) == null) // in ensureSegment
s = ensureSegment(j);
return s.put(key, hash, value, false);
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
主要注意的是 当前put 方法 当前key 为空的时候 ，代码报错。
这个代码主要是把Key 通过Hash函数计算出hash值 现计算出当前key属于那个Segment 调用Segment.put 分段方法Segment.put()
final V put(K key, int hash, V value, boolean onlyIfAbsent) {
HashEntry<K,V> node = tryLock() ? null :scanAndLockForPut(key, hash, value);
//tryLock()是ReentrantLock获取锁一个方法。如果当前线程获取锁成功 返回true,如果别线程获取了锁返回false不成功时会遍历定位到的HashEnry位置的链表（遍历主要是为了使CPU缓存链表），若找不到，则创建HashEntry。tryLock一定次数后（MAX_SCAN_RETRIES变量决定），则lock。若遍历过程中，由于其他线程的操作导致链表头结点变化，则需要重新遍历。
V oldValue;
try {
HashEntry<K,V>[] tab = table;
int index = (tab.length - 1) & hash;//定位HashEntry，可以看到，这个hash值在定位Segment时和在Segment中定位HashEntry都会用到，只不过定位Segment时只用到高几位。
HashEntry<K,V> first = entryAt(tab, index);
for (HashEntry<K,V> e = first;;) {
if (e != null) {
K k;
if ((k = e.key) == key ||
(e.hash == hash && key.equals(k))) {
oldValue = e.value;
if (!onlyIfAbsent) {
e.value = value;
++modCount;
}
break;
}
e = e.next;
}
else {
if (node != null)
node.setNext(first);
else
node = new HashEntry<K,V>(hash, key, value, first);
int c = count + 1;
//若c超出阈值threshold，需要扩容并rehash。扩容后的容量是当前容量的2倍。这样可以最大程度避免之前散列好的entry重新散列。扩容并rehash的这个过程是比较消耗资源的。
if (c > threshold && tab.length < MAXIMUM_CAPACITY)
rehash(node);
else
setEntryAt(tab, index, node);
++modCount;
count = c;
oldValue = null;
break;
}
}
} finally {
unlock();
}
return oldValue;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
Put 时候 ，通过Hash函数将即将要put 的元素均匀的放到所需要的Segment 段中，调用Segment的put 方法进行数据。
Segment的put 是加锁中完成的。如果当前元素数大于最大临界值的的话将会产生rehash. 先通过 getFirst 找到链表的表头部分，然后遍历链表，调用equals 比配是否存在相同的key ,如果找到的话，则将最新的Key 对应value值。如果没有找到，新增一个HashEntry 它加到整个Segment的头部。
我们先看一下Get 方法的源码：
//计算Segment中元素的数量
transient volatile int count;
***********************************************************
public V get(Object key) {
int hash = hash(key.hashCode());
return segmentFor(hash).get(key, hash);
}
***********************************************************
final Segment<K,V> segmentFor(int hash) {
return segments[(hash >>> segmentShift) & segmentMask];
}
********************************************************
V get(Object key, int hash) {
if (count != 0) { // read-volatile
HashEntry<K,V> e = getFirst(hash);
while (e != null) {
if (e.hash == hash && key.equals(e.key)) {
V v = e.value;
if (v != null)
return v;
return readValueUnderLock(e); // recheck
}
e = e.next;
}
}
return null;
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
1.读取的时候 传递Key值，通过Hash函数计算出 对应Segment 的位置。
2.调用segmentFor（int hash） 方法，用于确定操作应该在哪一个segment中进行 ，通过 右无符号位运算 右移segmentShift位在与运算 segmentMask【偏移码】 获得需要操作的Segment
确定了需要操作的Segment 再调用 get 方法获取对应的值。通过count 值先判断当前值是否为空。在调用getFirst（）获取头节点，然后遍历列表通过equals对比的方式进行比对返回值。
ConcurrentHashMap为什么读的时候不加锁？
ConcurrentHashMap是分段并发分段进行读取数据的。
Segment 里面有一个Count 字段，用来表示当前Segment中元素的个数 它的类型是volatile变量。所有的操作到最后都会 在最后一部更新count 这个变量，由于volatile变量 happer-before的特性。导致get 方法能够几乎准确的获取最新的结构更新。
再看一下ConcurrentHashMapRemove()方法：
V remove(Object key, int hash, Object value) {
lock();
try {
int c = count - 1;
HashEntry<K,V>[] tab = table;
int index = hash & (tab.length - 1);
HashEntry<K,V> first = tab[index];
HashEntry<K,V> e = first;
while (e != null && (e.hash != hash || !key.equals(e.key)))
e = e.next;
V oldValue = null;
if (e != null) {
V v = e.value;
if (value == null || value.equals(v)) {
oldValue = v;
// All entries following removed node can stay
// in list, but all preceding ones need to be
// cloned.
++modCount;
HashEntry<K,V> newFirst = e.next;
for (HashEntry<K,V> p = first; p != e; p = p.next)
newFirst = new HashEntry<K,V>(p.key, p.hash,
newFirst, p.value);
tab[index] = newFirst;
count = c; // write-volatile
}
}
return oldValue;
} finally {
unlock();
}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
调用Segment 的remove 方法，先定位当前要删除的元素C，此时需要把A、B元素全部复制一遍，一个一个接入到D上。
remove 也是在加锁的情况下进行的。
volatile 变量
我们发现 对于CurrentHashMap而言的话，源码里面又很多地方都用到了这个变量。比如HashEntry 、value 、Segment元素个数Count。
volatile 属于JMM 模型中的一个词语。首先先简单说一下 Java内存模型中的 几个概念：
原子性：保证 Java内存模型中原子变量内存操作的。通常有 read、write、load、use、assign、store、lock、unlock等这些。
可见性：就是当一个线程对一个变量进行了修改，其他线程即可立即得到这个变量最新的修改数据。
有序性：如果在本线程内观察，所有操作都是有序的；如果在一个线程中观察另一个线程，所有操作都是无序的。
先行发生：happen-before 先行发生原则是指Java内存模型中定义的两项操作之间的依序关系，如果说操作A先行发生于操作B，其实就是说发生操作B之前.
传递性
volatile 变量 与普通变量的不同之处？
volatile 是有可见性，一定程度的有序性。
volatile 赋值的时候新值能够立即刷新到主内存中去，每次使用的时候能够立刻从内存中刷新。
做一个简单例子看一下 这个功能
public class VolatileTest{
int a=1;
int b=2;
//赋值操作
public void change(){
a=3;
b=a;
}
//打印操作
public void print(){
System.out.println("b:"+b+",a:"+a);
}
@Test
public void testNorMal(){
VolatileTest vt=new VolatileTest();
for (int i = 0; i < 100000; i++) {
new Thread(new Runnable() {
@Override
public void run() {
try {
Thread.sleep(100);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
vt.change();
}
}).start();
new Thread(new Runnable() {
@Override
public void run() {
try {
Thread.sleep(10);
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
vt.print();
}
}).start();
}
}
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
跑了 n 次会出现一条 b=3,a=1 的错误打印记录。这就是因为普通变量相比volatile 不存在 可见性。
---------------------
作者：技术小排骨
来源：CSDN
原文：https://blog.csdn.net/jjc120074203/article/details/78625433
版权声明：本文为博主原创文章，转载请附上博文链接！
标签：实现 mda 完成 length 并发 依次 before ble int()
原文地址：https://www.cnblogs.com/softidea/p/10261414.html
