并发集合容器源码学习

Posted 2020-11-25 fatmanhappycode

ConcurrentHashmap (jdk1.7 & 1.8)

 

参考

 

ConcurrentHashmap1.7

 

1. ConcurrentHashmap1.7 和 hashmap 基本一样，只不过分成16段（ConcurrencyLevel 并发数，默认16，也是段数），并且每个segment都继承了Reentranlock，可以调用segment继承到的lock和unlock方法。

 

2. 初始化

时初始化segments数组的第0位，每个segment的默认大小为 2，负载因子loadFactor是 0.75，得出初始阈值为 1.5，也就是插入第一个元素不会触发扩容，插入第二个会进行第一次扩容。

 

3. sshift，2的sshift次方为 ConcurrencyLevel，segmentShift（移位数）的值为 32 - 4 = 28，segmentMask（掩码）为 16 - 1 = 15，hash 是 32 位，无符号右移 segmentShift(28) 位，剩下高 4 位 ，然后和 segmentMask(15) 做一次与操作，也就是说 j 是 hash 值的高 4 位，也就是槽的数组下标 j 。

 

4. 插入

前会先初始化 ensureSegment 插入的槽，使用当前 segment[0] 处的数组长度和负载因子来初始化 segment[k]，为什么要用“当前”，因为 segment[0] 可能早就扩容过了，CAS初始化。

 

5.  在往某个 segment 中 put 的时候，首先会调用  node = tryLock() ? null : scanAndLockForPut(key, hash, value) 也就是说先进行一次 tryLock() 快速获取该 segment 的独占锁，如果失败，那么进入到 scanAndLockForPut 这个方法来获取锁。scanAndLockForPut 就是重试次数如果超过 MAX_SCAN_RETRIES（单核1多核64），那么不抢了，进入到阻塞队列等待锁。

 

6. 扩容: rehash

首先会建立一个newTable用来迁移数据，迁移成功后指向它即可。其次，扩容有个比较有意思的点，就是如果遇到链表，会比较得到最后一个hash值不等于新位置下标值的结点，把这个结点赋给lastRun后，把lastRun结点放入新位置，而指向lastRun结点的后继结点也会因此一并移动到新位置，而剩下的lastRun之前的结点会重新计算hash值并放入应放的位置上。如果lastRun后的结点够多，就节省了把结点克隆的开销。（Doug Lea 也说了，根据统计，如果使用默认的阈值，大约只有 1/6 的节点需要克隆。）

 

ConcurrentHashmap1.8

 

1. sizeCtl

sizeCtl  CAS设置为 -1（标志位）是在初始化， int n = (sc > 0) ? sc : DEFAULT_CAPACITY; ，默认容量DEFAULT_CAPACITY为16，但是如果(sc = sizeCtl) > 0，则按照sizeCtl建tab数组。

sizeCtl在有initialCapacity初始化参数时的值为 sizeCtl = (1.5 * initialCapacity + 1) 然后向上取最近的 2 的 n 次方（容量），如 initialCapacity 为 10，那么得到 sizeCtl 为 16，如果 initialCapacity 为 11，得到 sizeCtl 为 32，没初始化值则默认16。

扩容前会变成一个很大的负数，每多一个线程进去帮忙扩容，则+1（再怎么加也是负数）

初始化完成后，或扩容成功后，sizeCtl为0.75 * n（阈值）。

 

2. 链表长度大于8调用treeifyBin方法变红黑树，但是这个方法和 HashMap 中稍微有一点点不同，那就是它不是一定会进行红黑树转换，如果当前数组的长度小于 64，那么会选择进行数组扩容，而不是转换为红黑树。

 

3. 扩容X2，扩容过程中的数据迁移需要执行 1 次 transfer(tab, null) + 多次 transfer(tab, nt)。

 

4. stride 在单核下直接等于 n，多核模式下为 (n>>>3)/NCPU，最小值是 16，stride 可以理解为”步长“，即每个线程处理stride个任务。

 

5. 迁移过程和hashmap差不多，但是由于是多线程，所以为了确认哪个位置已经被迁移，需要一个ForwardingNode确认位置，这个也是一个Node，不过它的hash为MOVED(-1)

 

6. 在putVal()的最后，有一个addCount方法会进行检查是否需要扩容，这个方法同样也调用了transfer去迁移数据

 

CopyOnWriteArrayList

 

参考 

 

ConcurrentSkipListMap

 

参考

 

BlockingQueue

BlockingQueue 是一个先进先出的队列（Queue），为什么说是阻塞（Blocking）的呢？是因为 BlockingQueue 支持当获取队列元素但是队列为空时，会阻塞等待队列中有元素再返回；也支持添加元素时，如果队列已满，那么等到队列可以放入新元素时再放入。

 

ArrayBlockingQueue

ArrayBlockingQueue 实现并发同步的原理就是，读操作和写操作都需要获取到 AQS 独占锁才能进行操作。如果队列为空，这个时候读操作的线程进入到读线程队列排队，等待写线程写入新的元素，然后唤醒读线程队列的第一个等待线程。如果队列已满，这个时候写操作的线程进入到写线程队列排队，等待读线程将队列元素移除腾出空间，然后唤醒写线程队列的第一个等待线程。

 

LinkedBlockingQueue

和ArrayBlockingQueue差不多实现，无非就是把数组改成链表，但是有一点比较奇怪，这里采用双锁（双ReentrantLock）实现，效率有提升，但是ArrayBlockingQueue却使用单锁，感觉这里应该也可以用双锁实现。

 

SynchronousQueue

• 不聊细节，一个队列，如果队列是空的，或者队列中的节点和当前的线程操作类型一致（如当前操作是 put 操作，而队列中的元素也都是put线程）。这种情况下，将当前线程加入到等待队列即可。
• 如果队列中有等待节点，而且与当前操作可以匹配（如队列中都是take操作线程，而当前线程是put操作线程）。这种情况下，匹配等待队列的队头，出队，返回相应数据。

 

PriorityBlockingQueue

一个加了Reentrantlock，在方法里lock一下的PriorityQueue，实现也同样是堆。