一 网站建设的目的与意义,新型互联网项目代理,项目四网站建设内容,外贸网站布局以前写过介绍HashMap的文章#xff0c;文中提到过HashMap在put的时候#xff0c;插入的元素超过了容量#xff08;由负载因子决定#xff09;的范围就会触发扩容操作#xff0c;就是rehash#xff0c;这个会重新将原数组的内容重新hash到新的扩容数组中#xff0c;在多线…以前写过介绍HashMap的文章文中提到过HashMap在put的时候插入的元素超过了容量由负载因子决定的范围就会触发扩容操作就是rehash这个会重新将原数组的内容重新hash到新的扩容数组中在多线程的环境下存在同时其他的元素也在进行put操作如果hash值相同可能出现同时在同一数组下用链表表示造成闭环导致在get时会出现死循环所以HashMap是线程不安全的。 JDK1.7的实现 整个 ConcurrentHashMap 由一个个 Segment 组成Segment 代表”部分“或”一段“的意思所以很多地方都会将其描述为分段锁。注意行文中我很多地方用了“槽”来代表一个 segment。 简单理解就是ConcurrentHashMap 是一个 Segment 数组Segment 通过继承 ReentrantLock 来进行加锁所以每次需要加锁的操作锁住的是一个 segment这样只要保证每个 Segment 是线程安全的也就实现了全局的线程安全。 concurrencyLevel并行级别、并发数、Segment 数。默认是 16也就是说 ConcurrentHashMap 有 16 个 Segments所以理论上这个时候最多可以同时支持 16 个线程并发写只要它们的操作分别分布在不同的 Segment 上。这个值可以在初始化的时候设置为其他值但是一旦初始化以后它是不可以扩容的。 再具体到每个 Segment 内部其实每个 Segment 很像之前介绍的 HashMap不过它要保证线程安全所以处理起来要麻烦些。 初始化 initialCapacity初始容量这个值指的是整个 ConcurrentHashMap 的初始容量实际操作的时候需要平均分给每个 Segment。 loadFactor负载因子之前我们说了Segment 数组不可以扩容所以这个负载因子是给每个 Segment 内部使用的。 1 public ConcurrentHashMap(int initialCapacity,2 float loadFactor, int concurrencyLevel) {3 if (!(loadFactor 0) || initialCapacity 0 || concurrencyLevel 0)4 throw new IllegalArgumentException();5 if (concurrencyLevel MAX_SEGMENTS)6 concurrencyLevel MAX_SEGMENTS;7 // Find power-of-two sizes best matching arguments8 int sshift 0;9 int ssize 1;
10 // 计算并行级别 ssize因为要保持并行级别是 2 的 n 次方
11 while (ssize concurrencyLevel) {
12 sshift;
13 ssize 1;
14 }
15 // 我们这里先不要那么烧脑用默认值concurrencyLevel 为 16sshift 为 4
16 // 那么计算出 segmentShift 为 28segmentMask 为 15后面会用到这两个值
17 this.segmentShift 32 - sshift;
18 this.segmentMask ssize - 1;
19
20 if (initialCapacity MAXIMUM_CAPACITY)
21 initialCapacity MAXIMUM_CAPACITY;
22
23 // initialCapacity 是设置整个 map 初始的大小
24 // 这里根据 initialCapacity 计算 Segment 数组中每个位置可以分到的大小
25 // 如 initialCapacity 为 64那么每个 Segment 或称之为槽可以分到 4 个
26 int c initialCapacity / ssize;
27 if (c * ssize initialCapacity)
28 c;
29 // 默认 MIN_SEGMENT_TABLE_CAPACITY 是 2这个值也是有讲究的因为这样的话对于具体的槽上
30 // 插入一个元素不至于扩容插入第二个的时候才会扩容
31 int cap MIN_SEGMENT_TABLE_CAPACITY;
32 while (cap c)
33 cap 1;
34
35 // 创建 Segment 数组
36 // 并创建数组的第一个元素 segment[0]
37 SegmentK,V s0
38 new SegmentK,V(loadFactor, (int)(cap * loadFactor),
39 (HashEntryK,V[])new HashEntry[cap]);
40 SegmentK,V[] ss (SegmentK,V[])new Segment[ssize];
41 // 往数组写入 segment[0]
42 UNSAFE.putOrderedObject(ss, SBASE, s0); // ordered write of segments[0]
43 this.segments ss;
44 } 初始化完成我们得到了一个 Segment 数组。 我们就当是用 new ConcurrentHashMap() 无参构造函数进行初始化的那么初始化完成后 Segment 数组长度为 16不可以扩容Segment[i] 的默认大小为 2负载因子是 0.75得出初始阈值为 1.5也就是以后插入第一个元素不会触发扩容插入第二个会进行第一次扩容这里初始化了 segment[0]其他位置还是 null至于为什么要初始化 segment[0]后面的代码会介绍当前 segmentShift 的值为 32 - 4 28segmentMask 为 16 - 1 15姑且把它们简单翻译为移位数和掩码这两个值马上就会用到Segment 1 static class SegmentK,V extends ReentrantLock implements Serializable {
2
3 transient volatile HashEntryK,V[] table;
4
5 transient int count;
6
7 transient int modCount;
8
9 } 从上Segment的继承体系可以看出Segment实现了ReentrantLock,也就带有锁的功能,table使用volatile修饰保证了内存可见性。 put 过程分析 我们先看 put 的主流程对于其中的一些关键细节操作后面会进行详细介绍。 1 public V put(K key, V value) {2 SegmentK,V s;3 if (value null)4 throw new NullPointerException();5 // 1. 计算 key 的 hash 值6 int hash hash(key);7 // 2. 根据 hash 值找到 Segment 数组中的位置 j8 // hash 是 32 位无符号右移 segmentShift(28) 位剩下高 4 位9 // 然后和 segmentMask(15) 做一次与操作也就是说 j 是 hash 值的高 4 位也就是槽的数组下标
10 int j (hash segmentShift) segmentMask;
11 // 刚刚说了初始化的时候初始化了 segment[0]但是其他位置还是 null
12 // ensureSegment(j) 对 segment[j] 进行初始化
13 if ((s (SegmentK,V)UNSAFE.getObject // nonvolatile; recheck
14 (segments, (j SSHIFT) SBASE)) null) // in ensureSegment
15 s ensureSegment(j);
16 // 3. 插入新值到 槽 s 中
17 return s.put(key, hash, value, false);
18 } 初始化槽: ensureSegment ConcurrentHashMap 初始化的时候会初始化第一个槽 segment[0]对于其他槽来说在插入第一个值的时候进行初始化。 这里需要考虑并发因为很可能会有多个线程同时进来初始化同一个槽 segment[k]不过只要有一个成功了就可以。 1 private SegmentK,V ensureSegment(int k) {2 final SegmentK,V[] ss this.segments;3 long u (k SSHIFT) SBASE; // raw offset4 SegmentK,V seg;5 if ((seg (SegmentK,V)UNSAFE.getObjectVolatile(ss, u)) null) {6 // 这里看到为什么之前要初始化 segment[0] 了7 // 使用当前 segment[0] 处的数组长度和负载因子来初始化 segment[k]8 // 为什么要用“当前”因为 segment[0] 可能早就扩容过了9 SegmentK,V proto ss[0];
10 int cap proto.table.length;
11 float lf proto.loadFactor;
12 int threshold (int)(cap * lf);
13
14 // 初始化 segment[k] 内部的数组
15 HashEntryK,V[] tab (HashEntryK,V[])new HashEntry[cap];
16 if ((seg (SegmentK,V)UNSAFE.getObjectVolatile(ss, u))
17 null) { // 再次检查一遍该槽是否被其他线程初始化了。
18
19 SegmentK,V s new SegmentK,V(lf, threshold, tab);
20 // 使用 while 循环内部用 CAS当前线程成功设值或其他线程成功设值后退出,如果其他线程成功设置后这里获取到直接返回
21 while ((seg (SegmentK,V)UNSAFE.getObjectVolatile(ss, u))
22 null) {
23 if (UNSAFE.compareAndSwapObject(ss, u, null, seg s))
24 break;
25 }
26 }
27 }
28 return seg;
29 } 总的来说ensureSegment(int k) 比较简单对于并发操作使用 CAS 进行控制。 第一层很简单根据 hash 值很快就能找到相应的 Segment之后就是 Segment 内部的 put 操作了。 Segment 内部是由 数组链表 组成的。 1 final V put(K key, int hash, V value, boolean onlyIfAbsent) {2 // 在往该 segment 写入前需要先获取该 segment 的独占锁3 // 先看主流程后面还会具体介绍这部分内容4 HashEntryK,V node tryLock() ? null :5 scanAndLockForPut(key, hash, value);6 V oldValue;7 try {8 // 这个是 segment 内部的数组9 HashEntryK,V[] tab table;
10 // 再利用 hash 值求应该放置的数组下标
11 int index (tab.length - 1) hash;
12 // first 是数组该位置处的链表的表头
13 HashEntryK,V first entryAt(tab, index);
14
15 // 下面这串 for 循环虽然很长不过也很好理解想想该位置没有任何元素和已经存在一个链表这两种情况
16 for (HashEntryK,V e first;;) {
17 if (e ! null) {
18 K k;
19 if ((k e.key) key ||
20 (e.hash hash key.equals(k))) {
21 oldValue e.value;
22 if (!onlyIfAbsent) {
23 // 覆盖旧值
24 e.value value;
25 modCount;
26 }
27 break;
28 }
29 // 继续顺着链表走
30 e e.next;
31 }
32 else {
33 // node 到底是不是 null这个要看获取锁的过程不过和这里都没有关系。
34 // 如果不为 null那就直接将它设置为链表表头如果是null初始化并设置为链表表头。
35 if (node ! null)
36 node.setNext(first);
37 else
38 node new HashEntryK,V(hash, key, value, first);
39
40 int c count 1;
41 // 如果超过了该 segment 的阈值这个 segment 需要扩容
42 if (c threshold tab.length MAXIMUM_CAPACITY)
43 rehash(node); // 扩容后面也会具体分析
44 else
45 // 没有达到阈值将 node 放到数组 tab 的 index 位置
46 // 其实就是将新的节点设置成原链表的表头
47 setEntryAt(tab, index, node);
48 modCount;
49 count c;
50 oldValue null;
51 break;
52 }
53 }
54 } finally {
55 // 解锁
56 unlock();
57 }
58 return oldValue;
59 } 整体流程还是比较简单的由于有独占锁的保护所以 segment 内部的操作并不复杂。至于这里面的并发问题我们稍后再进行介绍。 到这里 put 操作就结束了接下来我们说一说其中几步关键的操作。 获取写入锁: scanAndLockForPut 前面我们看到在往某个 segment 中 put 的时候首先会调用 node tryLock() ? null : scanAndLockForPut(key, hash, value)也就是说先进行一次 tryLock() 快速获取该 segment 的独占锁如果失败那么进入到 scanAndLockForPut 这个方法来获取锁。 下面我们来具体分析这个方法中是怎么控制加锁的。 1 private HashEntryK,V scanAndLockForPut(K key, int hash, V value) {2 HashEntryK,V first entryForHash(this, hash);3 HashEntryK,V e first;4 HashEntryK,V node null;5 int retries -1; // negative while locating node6 7 // 循环获取锁8 while (!tryLock()) {9 HashEntryK,V f; // to recheck first below
10 if (retries 0) {
11 if (e null) {
12 if (node null) // speculatively create node
13 // 进到这里说明数组该位置的链表是空的没有任何元素
14 // 当然进到这里的另一个原因是 tryLock() 失败所以该槽存在并发不一定是该位置
15 node new HashEntryK,V(hash, key, value, null);
16 retries 0;
17 }
18 else if (key.equals(e.key))
19 retries 0;
20 else
21 // 顺着链表往下走
22 e e.next;
23 }
24 // 重试次数如果超过 MAX_SCAN_RETRIES单核1多核64那么不抢了进入到阻塞队列等待锁
25 // lock() 是阻塞方法直到获取锁后返回
26 else if (retries MAX_SCAN_RETRIES) {
27 lock();
28 break;
29 }
30 else if ((retries 1) 0
31 // 这个时候是有大问题了那就是有新的元素进到了链表成为了新的表头
32 // 所以这边的策略是相当于重新走一遍这个 scanAndLockForPut 方法
33 (f entryForHash(this, hash)) ! first) {
34 e first f; // re-traverse if entry changed
35 retries -1;
36 }
37 }
38 return node;
39 } 这个方法有两个出口一个是 tryLock() 成功了循环终止另一个就是重试次数超过了 MAX_SCAN_RETRIES进到 lock() 方法此方法会阻塞等待直到成功拿到独占锁。 这个方法就是看似复杂但是其实就是做了一件事那就是获取该 segment 的独占锁如果需要的话顺便实例化了一下 node。 获取锁时并不直接使用lock来获取因为该方法获取锁失败时会挂起。事实上它使用了自旋锁如果tryLock获取锁失败说明锁被其它线程占用此时通过循环再次以tryLock的方式申请锁。如果在循环过程中该Key所对应的链表头被修改则重置retry次数。如果retry次数超过一定值则使用lock方法申请锁。 这里使用自旋锁是因为自旋锁的效率比较高但是它消耗CPU资源比较多因此在自旋次数超过阈值时切换为互斥锁。 扩容: rehash 重复一下segment 数组不能扩容扩容是 segment 数组某个位置内部的数组 HashEntry\k,v[] 进行扩容扩容后容量为原来的 2 倍。 首先我们要回顾一下触发扩容的地方put 的时候如果判断该值的插入会导致该 segment 的元素个数超过阈值那么先进行扩容再插值读者这个时候可以回去 put 方法看一眼。 该方法不需要考虑并发因为到这里的时候是持有该 segment 的独占锁的。 1 // 方法参数上的 node 是这次扩容后需要添加到新的数组中的数据。2 private void rehash(HashEntryK,V node) {3 HashEntryK,V[] oldTable table;4 int oldCapacity oldTable.length;5 // 2 倍6 int newCapacity oldCapacity 1;7 threshold (int)(newCapacity * loadFactor);8 // 创建新数组9 HashEntryK,V[] newTable
10 (HashEntryK,V[]) new HashEntry[newCapacity];
11 // 新的掩码如从 16 扩容到 32那么 sizeMask 为 31对应二进制 ‘000...00011111’
12 int sizeMask newCapacity - 1;
13
14 // 遍历原数组老套路将原数组位置 i 处的链表拆分到 新数组位置 i 和 ioldCap 两个位置
15 for (int i 0; i oldCapacity ; i) {
16 // e 是链表的第一个元素
17 HashEntryK,V e oldTable[i];
18 if (e ! null) {
19 HashEntryK,V next e.next;
20 // 计算应该放置在新数组中的位置
21 // 假设原数组长度为 16e 在 oldTable[3] 处那么 idx 只可能是 3 或者是 3 16 19
22 int idx e.hash sizeMask;
23 if (next null) // 该位置处只有一个元素那比较好办
24 newTable[idx] e;
25 else { // Reuse consecutive sequence at same slot
26 // e 是链表表头
27 HashEntryK,V lastRun e;
28 // idx 是当前链表的头结点 e 的新位置
29 int lastIdx idx;
30
31 // 下面这个 for 循环会找到一个 lastRun 节点这个节点之后的所有元素是将要放到一起的
32 for (HashEntryK,V last next;
33 last ! null;
34 last last.next) {
35 int k last.hash sizeMask;
36 if (k ! lastIdx) {
37 lastIdx k;
38 lastRun last;
39 }
40 }
41 // 将 lastRun 及其之后的所有节点组成的这个链表放到 lastIdx 这个位置
42 newTable[lastIdx] lastRun;
43 // 下面的操作是处理 lastRun 之前的节点
44 // 这些节点可能分配在另一个链表中也可能分配到上面的那个链表中
45 for (HashEntryK,V p e; p ! lastRun; p p.next) {
46 V v p.value;
47 int h p.hash;
48 int k h sizeMask;
49 HashEntryK,V n newTable[k];
50 newTable[k] new HashEntryK,V(h, p.key, v, n);
51 }
52 }
53 }
54 }
55 // 将新来的 node 放到新数组中刚刚的 两个链表之一 的 头部
56 int nodeIndex node.hash sizeMask; // add the new node
57 node.setNext(newTable[nodeIndex]);
58 newTable[nodeIndex] node;
59 table newTable;
60 } 总结一下put的流程 当执行put操作时会进行第一次key的hash来定位Segment的位置如果该Segment还没有初始化即通过CAS操作进行赋值然后进行第二次hash操作找到相应的HashEntry的位置这里会利用继承过来的锁的特性在将数据插入指定的HashEntry位置时链表的尾端会通过继承ReentrantLock的tryLock方法尝试去获取锁如果获取成功就直接插入相应的位置如果已经有线程获取该Segment的锁那当前线程会以自旋的方式去继续的调用tryLock方法去获取锁超过指定次数就挂起等待唤醒。 get 过程分析 相对于 put 来说get 真的不要太简单。 计算 hash 值找到 segment 数组中的具体位置或我们前面用的“槽”槽中也是一个数组根据 hash 找到数组中具体的位置到这里是链表了顺着链表进行查找即可 1 public V get(Object key) {2 SegmentK,V s; // manually integrate access methods to reduce overhead3 HashEntryK,V[] tab;4 // 1. hash 值5 int h hash(key);6 long u (((h segmentShift) segmentMask) SSHIFT) SBASE;7 // 2. 根据 hash 找到对应的 segment8 if ((s (SegmentK,V)UNSAFE.getObjectVolatile(segments, u)) ! null 9 (tab s.table) ! null) {
10 // 3. 找到segment 内部数组相应位置的链表遍历
11 for (HashEntryK,V e (HashEntryK,V) UNSAFE.getObjectVolatile
12 (tab, ((long)(((tab.length - 1) h)) TSHIFT) TBASE);
13 e ! null; e e.next) {
14 K k;
15 if ((k e.key) key || (e.hash h key.equals(k)))
16 return e.value;
17 }
18 }
19 return null;
20 } size操作 put、remove和get操作只需要关心一个Segment而size操作需要遍历所有的Segment才能算出整个Map的大小。一个简单的方案是先锁住所有Sgment计算完后再解锁。但这样做在做size操作时不仅无法对Map进行写操作同时也无法进行读操作不利于对Map的并行操作。 为更好支持并发操作ConcurrentHashMap会在不上锁的前提逐个Segment计算3次size如果某相邻两次计算获取的所有Segment的更新次数每个Segment都与HashMap一样通过modCount跟踪自己的修改次数Segment每修改一次其modCount加一相等说明这两次计算过程中无更新操作则这两次计算出的总size相等可直接作为最终结果返回。如果这三次计算过程中Map有更新则对所有Segment加锁重新计算Size。该计算方法代码如下 1 public int size() {2 final SegmentK,V[] segments this.segments;3 int size;4 boolean overflow; // true if size overflows 32 bits5 long sum; // sum of modCounts6 long last 0L; // previous sum7 int retries -1; // first iteration isnt retry8 try {9 for (;;) {
10 if (retries RETRIES_BEFORE_LOCK) {
11 for (int j 0; j segments.length; j)
12 ensureSegment(j).lock(); // force creation
13 }
14 sum 0L;
15 size 0;
16 overflow false;
17 for (int j 0; j segments.length; j) {
18 SegmentK,V seg segmentAt(segments, j);
19 if (seg ! null) {
20 sum seg.modCount;
21 int c seg.count;
22 if (c 0 || (size c) 0)
23 overflow true;
24 }
25 }
26 if (sum last)
27 break;
28 last sum;
29 }
30 } finally {
31 if (retries RETRIES_BEFORE_LOCK) {
32 for (int j 0; j segments.length; j)
33 segmentAt(segments, j).unlock();
34 }
35 }
36 return overflow ? Integer.MAX_VALUE : size;
37 } ConcurrentHashMap的Size方法是一个嵌套循环大体逻辑如下 1.遍历所有的Segment。 2.把Segment的元素数量累加起来。 3.把Segment的修改次数累加起来。 4.判断所有Segment的总修改次数是否大于上一次的总修改次数。如果大于说明统计过程中有修改重新统计尝试次数1如果不是。说明没有修改统计结束。 5.如果尝试次数超过阈值则对每一个Segment加锁再重新统计。 6.再次判断所有Segment的总修改次数是否大于上一次的总修改次数。由于已经加锁次数一定和上次相等。 7.释放锁统计结束。 并发问题分析 现在我们已经说完了 put 过程和 get 过程我们可以看到 get 过程中是没有加锁的那自然我们就需要去考虑并发问题。 添加节点的操作 put 和删除节点的操作 remove 都是要加 segment 上的独占锁的所以它们之间自然不会有问题我们需要考虑的问题就是 get 的时候在同一个 segment 中发生了 put 或 remove 操作。 put 操作的线程安全性。 初始化槽这个我们之前就说过了使用了 CAS 来初始化 Segment 中的数组。添加节点到链表的操作是插入到表头的所以如果这个时候 get 操作在链表遍历的过程已经到了中间是不会影响的。当然另一个并发问题就是 get 操作在 put 之后需要保证刚刚插入表头的节点被读取这个依赖于 setEntryAt 方法中使用的 UNSAFE.putOrderedObject。扩容。扩容是新创建了数组然后进行迁移数据最后面将 newTable 设置给属性 table。所以如果 get 操作此时也在进行那么也没关系如果 get 先行那么就是在旧的 table 上做查询操作而 put 先行那么 put 操作的可见性保证就是 table 使用了 volatile 关键字。 remove 操作的线程安全性。 remove 操作我们没有分析源码所以这里说的读者感兴趣的话还是需要到源码中去求实一下的。 get 操作需要遍历链表但是 remove 操作会破坏链表。 如果 remove 破坏的节点 get 操作已经过去了那么这里不存在任何问题。 如果 remove 先破坏了一个节点分两种情况考虑。 1、如果此节点是头结点那么需要将头结点的 next 设置为数组该位置的元素table 虽然使用了 volatile 修饰但是 volatile 并不能提供数组内部操作的可见性保证所以源码中使用了 UNSAFE 来操作数组请看方法 setEntryAt。2、如果要删除的节点不是头结点它会将要删除节点的后继节点接到前驱节点中这里的并发保证就是 next 属性是 volatile 的。 推荐博客 https://www.cnblogs.com/chen-haozi/p/10227797.html 最后我们来看看并发操作示意图 Case1不同Segment的并发写入 不同Segment的写入是可以并发执行的。 Case2同一Segment的一写一读 同一Segment的写和读是可以并发执行的。 Case3同一Segment的并发写入 Segment的写入是需要上锁的因此对同一Segment的并发写入会被阻塞。 由此可见ConcurrentHashMap当中每个Segment各自持有一把锁。在保证线程安全的同时降低了锁的粒度让并发操作效率更高。 转载于:https://www.cnblogs.com/java-chen-hao/p/10327280.html
