From ab8df710b2eb28f9859a23703b6ede710cf3ef24 Mon Sep 17 00:00:00 2001 From: dbcxyxd <43689452+dbcxyxd@users.noreply.github.com> Date: Fri, 3 Jan 2020 21:02:37 +0800 Subject: [PATCH] week_51 --- week_04/51/ConcurrentHashMap_51.md | 987 ++++++++++++++++++ week_04/51/ConcurrentLinkedQueue_51.md | 217 ++++ week_04/51/CopyOnWriteArrayList_51.md | 355 +++++++ .../51/Week4_51\346\200\273\347\273\223.md" | 75 ++ 4 files changed, 1634 insertions(+) create mode 100644 week_04/51/ConcurrentHashMap_51.md create mode 100644 week_04/51/ConcurrentLinkedQueue_51.md create mode 100644 week_04/51/CopyOnWriteArrayList_51.md create mode 100644 "week_04/51/Week4_51\346\200\273\347\273\223.md" diff --git a/week_04/51/ConcurrentHashMap_51.md b/week_04/51/ConcurrentHashMap_51.md new file mode 100644 index 0000000..9390dcb --- /dev/null +++ b/week_04/51/ConcurrentHashMap_51.md @@ -0,0 +1,987 @@ +## 开篇问题 + +（1）ConcurrentHashMap与HashMap的数据结构是否一样？ + +（2）HashMap在多线程环境下何时会出现并发安全问题？ + +（3）ConcurrentHashMap是怎么解决并发安全问题的？ + +（4）ConcurrentHashMap使用了哪些锁？ + +（5）ConcurrentHashMap的扩容是怎么进行的？ + +（6）ConcurrentHashMap是否是强一致性的？ + +（7）ConcurrentHashMap不能解决哪些问题？ + +（8）ConcurrentHashMap中有哪些不常见的技术值得学习？ + +## 简介 + +ConcurrentHashMap是HashMap的线程安全版本，内部也是使用（数组 + 链表 + 红黑树）的结构来存储元素。 + +相比于同样线程安全的HashTable来说，效率等各方面都有极大地提高。 + +## 各种锁简介 + +这里先简单介绍一下各种锁，以便下文讲到相关概念时能有个印象。 + +（1）synchronized + +java中的关键字，内部实现为监视器锁，主要是通过对象监视器在对象头中的字段来表明的。 + +synchronized从旧版本到现在已经做了很多优化了，在运行时会有三种存在方式：偏向锁，轻量级锁，重量级锁。 + +偏向锁，是指一段同步代码一直被一个线程访问，那么这个线程会自动获取锁，降低获取锁的代价。 + +轻量级锁，是指当锁是偏向锁时，被另一个线程所访问，偏向锁会升级为轻量级锁，这个线程会通过自旋的方式尝试获取锁，不会阻塞，提高性能。 + +重量级锁，是指当锁是轻量级锁时，当自旋的线程自旋了一定的次数后，还没有获取到锁，就会进入阻塞状态，该锁升级为重量级锁，重量级锁会使其他线程阻塞，性能降低。 + +（2）CAS + +CAS，Compare And Swap，它是一种乐观锁，认为对于同一个数据的并发操作不一定会发生修改，在更新数据的时候，尝试去更新数据，如果失败就不断尝试。 + +（3）volatile（非锁） + +java中的关键字，当多个线程访问同一个变量时，一个线程修改了这个变量的值，其他线程能够立即看得到修改的值。（这里牵涉到java内存模型的知识，感兴趣的同学可以自己查查相关资料） + +volatile只保证可见性，不保证原子性，比如 volatile修改的变量 i，针对i++操作，不保证每次结果都正确，因为i++操作是两步操作，相当于 i = i +1，先读取，再加1，这种情况volatile是无法保证的。 + +（4）自旋锁 + +自旋锁，是指尝试获取锁的线程不会阻塞，而是循环的方式不断尝试，这样的好处是减少线程的上下文切换带来的开锁，提高性能，缺点是循环会消耗CPU。 + +（5）分段锁 + +分段锁，是一种锁的设计思路，它细化了锁的粒度，主要运用在ConcurrentHashMap中，实现高效的并发操作，当操作不需要更新整个数组时，就只锁数组中的一项就可以了。 + +（5）ReentrantLock + +可重入锁，是指一个线程获取锁之后再尝试获取锁时会自动获取锁，可重入锁的优点是避免死锁。 + +其实，synchronized也是可重入锁。 + +## 源码分析 + +### 构造方法 + +``` +public ConcurrentHashMap() { +} + +public ConcurrentHashMap(int initialCapacity) { + if (initialCapacity < 0) + throw new IllegalArgumentException(); + int cap = ((initialCapacity >= (MAXIMUM_CAPACITY >>> 1)) ? + MAXIMUM_CAPACITY : + tableSizeFor(initialCapacity + (initialCapacity >>> 1) + 1)); + this.sizeCtl = cap; +} + +public ConcurrentHashMap(Map m) { + this.sizeCtl = DEFAULT_CAPACITY; + putAll(m); +} + +public ConcurrentHashMap(int initialCapacity, float loadFactor) { + this(initialCapacity, loadFactor, 1); +} + +public ConcurrentHashMap(int initialCapacity, + float loadFactor, int concurrencyLevel) { + if (!(loadFactor > 0.0f) || initialCapacity < 0 || concurrencyLevel <= 0) + throw new IllegalArgumentException(); + if (initialCapacity < concurrencyLevel) // Use at least as many bins + initialCapacity = concurrencyLevel; // as estimated threads + long size = (long)(1.0 + (long)initialCapacity / loadFactor); + int cap = (size >= (long)MAXIMUM_CAPACITY) ? + MAXIMUM_CAPACITY : tableSizeFor((int)size); + this.sizeCtl = cap; +} +``` + +构造方法与HashMap对比可以发现，没有了HashMap中的threshold和loadFactor，而是改用了sizeCtl来控制，而且只存储了容量在里面，那么它是怎么用的呢？官方给出的解释如下： + +（1）-1，表示有线程正在进行初始化操作 + +（2）-(1 + nThreads)，表示有n个线程正在一起扩容 + +（3）0，默认值，后续在真正初始化的时候使用默认容量 + +（4）> 0，初始化或扩容完成后下一次的扩容门槛 + +至于，官方这个解释对不对我们后面再讨论。 + +### 添加元素 + +``` +public V put(K key, V value) { + return putVal(key, value, false); +} + +final V putVal(K key, V value, boolean onlyIfAbsent) { + // key和value都不能为null + if (key == null || value == null) throw new NullPointerException(); + // 计算hash值 + int hash = spread(key.hashCode()); + // 要插入的元素所在桶的元素个数 + int binCount = 0; + // 死循环，结合CAS使用（如果CAS失败，则会重新取整个桶进行下面的流程） + for (Node[] tab = table;;) { + Node f; int n, i, fh; + if (tab == null || (n = tab.length) == 0) + // 如果桶未初始化或者桶个数为0，则初始化桶 + tab = initTable(); + else if ((f = tabAt(tab, i = (n - 1) & hash)) == null) { + // 如果要插入的元素所在的桶还没有元素，则把这个元素插入到这个桶中 + if (casTabAt(tab, i, null, + new Node(hash, key, value, null))) + // 如果使用CAS插入元素时，发现已经有元素了，则进入下一次循环，重新操作 + // 如果使用CAS插入元素成功，则break跳出循环，流程结束 + break; // no lock when adding to empty bin + } + else if ((fh = f.hash) == MOVED) + // 如果要插入的元素所在的桶的第一个元素的hash是MOVED，则当前线程帮忙一起迁移元素 + tab = helpTransfer(tab, f); + else { + // 如果这个桶不为空且不在迁移元素，则锁住这个桶（分段锁） + // 并查找要插入的元素是否在这个桶中 + // 存在，则替换值（onlyIfAbsent=false） + // 不存在，则插入到链表结尾或插入树中 + V oldVal = null; + synchronized (f) { + // 再次检测第一个元素是否有变化，如果有变化则进入下一次循环，从头来过 + if (tabAt(tab, i) == f) { + // 如果第一个元素的hash值大于等于0（说明不是在迁移，也不是树） + // 那就是桶中的元素使用的是链表方式存储 + if (fh >= 0) { + // 桶中元素个数赋值为1 + binCount = 1; + // 遍历整个桶，每次结束binCount加1 + for (Node e = f;; ++binCount) { + K ek; + if (e.hash == hash && + ((ek = e.key) == key || + (ek != null && key.equals(ek)))) { + // 如果找到了这个元素，则赋值了新值（onlyIfAbsent=false） + // 并退出循环 + oldVal = e.val; + if (!onlyIfAbsent) + e.val = value; + break; + } + Node pred = e; + if ((e = e.next) == null) { + // 如果到链表尾部还没有找到元素 + // 就把它插入到链表结尾并退出循环 + pred.next = new Node(hash, key, + value, null); + break; + } + } + } + else if (f instanceof TreeBin) { + // 如果第一个元素是树节点 + Node p; + // 桶中元素个数赋值为2 + binCount = 2; + // 调用红黑树的插入方法插入元素 + // 如果成功插入则返回null + // 否则返回寻找到的节点 + if ((p = ((TreeBin)f).putTreeVal(hash, key, + value)) != null) { + // 如果找到了这个元素，则赋值了新值（onlyIfAbsent=false） + // 并退出循环 + oldVal = p.val; + if (!onlyIfAbsent) + p.val = value; + } + } + } + } + // 如果binCount不为0，说明成功插入了元素或者寻找到了元素 + if (binCount != 0) { + // 如果链表元素个数达到了8，则尝试树化 + // 因为上面把元素插入到树中时，binCount只赋值了2，并没有计算整个树中元素的个数 + // 所以不会重复树化 + if (binCount >= TREEIFY_THRESHOLD) + treeifyBin(tab, i); + // 如果要插入的元素已经存在，则返回旧值 + if (oldVal != null) + return oldVal; + // 退出外层大循环，流程结束 + break; + } + } + } + // 成功插入元素，元素个数加1（是否要扩容在这个里面） + addCount(1L, binCount); + // 成功插入元素返回null + return null; + } +``` + +整体流程跟HashMap比较类似，大致是以下几步： + +（1）如果桶数组未初始化，则初始化； + +（2）如果待插入的元素所在的桶为空，则尝试把此元素直接插入到桶的第一个位置； + +（3）如果正在扩容，则当前线程一起加入到扩容的过程中； + +（4）如果待插入的元素所在的桶不为空且不在迁移元素，则锁住这个桶（分段锁）； + +（5）如果当前桶中元素以链表方式存储，则在链表中寻找该元素或者插入元素； + +（6）如果当前桶中元素以红黑树方式存储，则在红黑树中寻找该元素或者插入元素； + +（7）如果元素存在，则返回旧值； + +（8）如果元素不存在，整个Map的元素个数加1，并检查是否需要扩容； + +添加元素操作中使用的锁主要有（自旋锁 + CAS + synchronized + 分段锁）。 + +为什么使用synchronized而不是ReentrantLock？ + +因为synchronized已经得到了极大地优化，在特定情况下并不比ReentrantLock差。 + + + +### 初始化桶数组 + +第一次放元素时，初始化桶数组。 + +``` +private final Node[] initTable() { + Node[] tab; int sc; + while ((tab = table) == null || tab.length == 0) { + if ((sc = sizeCtl) < 0) + // 如果sizeCtl<0说明正在初始化或者扩容，让出CPU + Thread.yield(); // lost initialization race; just spin + else if (U.compareAndSwapInt(this, SIZECTL, sc, -1)) { + // 如果把sizeCtl原子更新为-1成功，则当前线程进入初始化 + // 如果原子更新失败则说明有其它线程先一步进入初始化了，则进入下一次循环 + // 如果下一次循环时还没初始化完毕，则sizeCtl<0进入上面if的逻辑让出CPU + // 如果下一次循环更新完毕了，则table.length!=0，退出循环 + try { + // 再次检查table是否为空，防止ABA问题 + if ((tab = table) == null || tab.length == 0) { + // 如果sc为0则使用默认值16 + int n = (sc > 0) ? sc : DEFAULT_CAPACITY; + // 新建数组 + @SuppressWarnings("unchecked") + Node[] nt = (Node[])new Node[n]; + // 赋值给table桶数组 + table = tab = nt; + // 设置sc为数组长度的0.75倍 + // n - (n >>> 2) = n - n/4 = 0.75n + // 可见这里装载因子和扩容门槛都是写死了的 + // 这也正是没有threshold和loadFactor属性的原因 + sc = n - (n >>> 2); + } + } finally { + // 把sc赋值给sizeCtl，这时存储的是扩容门槛 + sizeCtl = sc; + } + break; + } + } + return tab; +} +``` + +（1）使用CAS锁控制只有一个线程初始化桶数组； + +（2）sizeCtl在初始化后存储的是扩容门槛； + +（3）扩容门槛写死的是桶数组大小的0.75倍，桶数组大小即map的容量，也就是最多存储多少个元素。 + +### 判断是否需要扩容 + +每次添加元素后，元素数量加1，并判断是否达到扩容门槛，达到了则进行扩容或协助扩容。 + +``` +private final void addCount(long x, int check) { + CounterCell[] as; long b, s; + // 这里使用的思想跟LongAdder类是一模一样的（后面会讲） + // 把数组的大小存储根据不同的线程存储到不同的段上（也是分段锁的思想） + // 并且有一个baseCount，优先更新baseCount，如果失败了再更新不同线程对应的段 + // 这样可以保证尽量小的减少冲突 + + // 先尝试把数量加到baseCount上，如果失败再加到分段的CounterCell上 + if ((as = counterCells) != null || + !U.compareAndSwapLong(this, BASECOUNT, b = baseCount, s = b + x)) { + CounterCell a; long v; int m; + boolean uncontended = true; + // 如果as为空 + // 或者长度为0 + // 或者当前线程所在的段为null + // 或者在当前线程的段上加数量失败 + if (as == null || (m = as.length - 1) < 0 || + (a = as[ThreadLocalRandom.getProbe() & m]) == null || + !(uncontended = + U.compareAndSwapLong(a, CELLVALUE, v = a.value, v + x))) { + // 强制增加数量（无论如何数量是一定要加上的，并不是简单地自旋） + // 不同线程对应不同的段都更新失败了 + // 说明已经发生冲突了，那么就对counterCells进行扩容 + // 以减少多个线程hash到同一个段的概率 + fullAddCount(x, uncontended); + return; + } + if (check <= 1) + return; + // 计算元素个数 + s = sumCount(); + } + if (check >= 0) { + Node[] tab, nt; int n, sc; + // 如果元素个数达到了扩容门槛，则进行扩容 + // 注意，正常情况下sizeCtl存储的是扩容门槛，即容量的0.75倍 + while (s >= (long)(sc = sizeCtl) && (tab = table) != null && + (n = tab.length) < MAXIMUM_CAPACITY) { + // rs是扩容时的一个邮戳标识 + int rs = resizeStamp(n); + if (sc < 0) { + // sc<0说明正在扩容中 + if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 || + sc == rs + MAX_RESIZERS || (nt = nextTable) == null || + transferIndex <= 0) + // 扩容已经完成了，退出循环 + // 正常应该只会触发nextTable==null这个条件，其它条件没看出来何时触发 + break; + + // 扩容未完成，则当前线程加入迁移元素中 + // 并把扩容线程数加1 + if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) + transfer(tab, nt); + } + else if (U.compareAndSwapInt(this, SIZECTL, sc, + (rs << RESIZE_STAMP_SHIFT) + 2)) + // 这里是触发扩容的那个线程进入的地方 + // sizeCtl的高16位存储着rs这个扩容邮戳 + // sizeCtl的低16位存储着扩容线程数加1，即(1+nThreads) + // 所以官方说的扩容时sizeCtl的值为 -(1+nThreads)是错误的 + + // 进入迁移元素 + transfer(tab, null); + // 重新计算元素个数 + s = sumCount(); + } + } +} +``` + +（1）元素个数的存储方式类似于LongAdder类，存储在不同的段上，减少不同线程同时更新size时的冲突； + +（2）计算元素个数时把这些段的值及baseCount相加算出总的元素个数； + +（3）正常情况下sizeCtl存储着扩容门槛，扩容门槛为容量的0.75倍； + +（4）扩容时sizeCtl高位存储扩容邮戳(resizeStamp)，低位存储扩容线程数加1（1+nThreads）； + +（5）其它线程添加元素后如果发现存在扩容，也会加入的扩容行列中来； + +### 协助扩容（迁移元素） + +线程添加元素时发现正在扩容且当前元素所在的桶元素已经迁移完成了，则协助迁移其它桶的元素。 + +``` +final Node[] helpTransfer(Node[] tab, Node f) { + Node[] nextTab; int sc; + // 如果桶数组不为空，并且当前桶第一个元素为ForwardingNode类型，并且nextTab不为空 + // 说明当前桶已经迁移完毕了，才去帮忙迁移其它桶的元素 + // 扩容时会把旧桶的第一个元素置为ForwardingNode，并让其nextTab指向新桶数组 + if (tab != null && (f instanceof ForwardingNode) && + (nextTab = ((ForwardingNode)f).nextTable) != null) { + int rs = resizeStamp(tab.length); + // sizeCtl<0，说明正在扩容 + while (nextTab == nextTable && table == tab && + (sc = sizeCtl) < 0) { + if ((sc >>> RESIZE_STAMP_SHIFT) != rs || sc == rs + 1 || + sc == rs + MAX_RESIZERS || transferIndex <= 0) + break; + // 扩容线程数加1 + if (U.compareAndSwapInt(this, SIZECTL, sc, sc + 1)) { + // 当前线程帮忙迁移元素 + transfer(tab, nextTab); + break; + } + } + return nextTab; + } + return table; +} +``` + +当前桶元素迁移完成了才去协助迁移其它桶元素； + +### 迁移元素 + +扩容时容量变为两倍，并把部分元素迁移到其它桶中。 + +``` +private final void transfer(Node[] tab, Node[] nextTab) { + int n = tab.length, stride; + if ((stride = (NCPU > 1) ? (n >>> 3) / NCPU : n) < MIN_TRANSFER_STRIDE) + stride = MIN_TRANSFER_STRIDE; // subdivide range + if (nextTab == null) { // initiating + // 如果nextTab为空，说明还没开始迁移 + // 就新建一个新桶数组 + try { + // 新桶数组是原桶的两倍 + @SuppressWarnings("unchecked") + Node[] nt = (Node[])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类型的节点，并把新桶数组存储在里面 + ForwardingNode fwd = new ForwardingNode(nextTab); + boolean advance = true; + boolean finishing = false; // to ensure sweep before committing nextTab + for (int i = 0, bound = 0;;) { + Node f; int fh; + // 整个while循环就是在算i的值，过程太复杂，不用太关心 + // i的值会从n-1依次递减，感兴趣的可以打下断点就知道了 + // 其中n是旧桶数组的大小，也就是说i从15开始一直减到1这样去迁移元素 + while (advance) { + int nextIndex, nextBound; + if (--i >= bound || finishing) + advance = false; + else if ((nextIndex = transferIndex) <= 0) { + i = -1; + advance = false; + } + 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) { + // 如果一次遍历完成了 + // 也就是整个map所有桶中的元素都迁移完成了 + int sc; + if (finishing) { + // 如果全部迁移完成了，则替换旧桶数组 + // 并设置下一次扩容门槛为新桶数组容量的0.75倍 + nextTable = null; + table = nextTab; + sizeCtl = (n << 1) - (n >>> 1); + return; + } + if (U.compareAndSwapInt(this, SIZECTL, sc = sizeCtl, sc - 1)) { + // 当前线程扩容完成，把扩容线程数-1 + if ((sc - 2) != resizeStamp(n) << RESIZE_STAMP_SHIFT) + // 扩容完成两边肯定相等 + return; + // 把finishing设置为true + // finishing为true才会走到上面的if条件 + finishing = advance = true; + // i重新赋值为n + // 这样会再重新遍历一次桶数组，看看是不是都迁移完成了 + // 也就是第二次遍历都会走到下面的(fh = f.hash) == MOVED这个条件 + i = n; // recheck before commit + } + } + else if ((f = tabAt(tab, i)) == null) + // 如果桶中无数据，直接放入ForwardingNode标记该桶已迁移 + advance = casTabAt(tab, i, null, fwd); + else if ((fh = f.hash) == MOVED) + // 如果桶中第一个元素的hash值为MOVED + // 说明它是ForwardingNode节点 + // 也就是该桶已迁移 + advance = true; // already processed + else { + // 锁定该桶并迁移元素 + synchronized (f) { + // 再次判断当前桶第一个元素是否有修改 + // 也就是可能其它线程先一步迁移了元素 + if (tabAt(tab, i) == f) { + // 把一个链表分化成两个链表 + // 规则是桶中各元素的hash与桶大小n进行与操作 + // 等于0的放到低位链表(low)中，不等于0的放到高位链表(high)中 + // 其中低位链表迁移到新桶中的位置相对旧桶不变 + // 高位链表迁移到新桶中位置正好是其在旧桶的位置加n + // 这也正是为什么扩容时容量在变成两倍的原因 + Node ln, hn; + if (fh >= 0) { + // 第一个元素的hash值大于等于0 + // 说明该桶中元素是以链表形式存储的 + // 这里与HashMap迁移算法基本类似 + // 唯一不同的是多了一步寻找lastRun + // 这里的lastRun是提取出链表后面不用处理再特殊处理的子链表 + // 比如所有元素的hash值与桶大小n与操作后的值分别为 0 0 4 4 0 0 0 + // 则最后后面三个0对应的元素肯定还是在同一个桶中 + // 这时lastRun对应的就是倒数第三个节点 + // 至于为啥要这样处理，我也没太搞明白 + int runBit = fh & n; + Node lastRun = f; + for (Node 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; + } + // 遍历链表，把hash&n为0的放在低位链表中 + // 不为0的放在高位链表中 + for (Node 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(ph, pk, pv, ln); + else + hn = new Node(ph, pk, pv, hn); + } + // 低位链表的位置不变 + setTabAt(nextTab, i, ln); + // 高位链表的位置是原位置加n + setTabAt(nextTab, i + n, hn); + // 标记当前桶已迁移 + setTabAt(tab, i, fwd); + // advance为true，返回上面进行--i操作 + advance = true; + } + else if (f instanceof TreeBin) { + // 如果第一个元素是树节点 + // 也是一样，分化成两颗树 + // 也是根据hash&n为0放在低位树中 + // 不为0放在高位树中 + TreeBin t = (TreeBin)f; + TreeNode lo = null, loTail = null; + TreeNode hi = null, hiTail = null; + int lc = 0, hc = 0; + // 遍历整颗树，根据hash&n是否为0分化成两颗树 + for (Node e = t.first; e != null; e = e.next) { + int h = e.hash; + TreeNode p = new TreeNode + (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; + } + } + // 如果分化的树中元素个数小于等于6，则退化成链表 + ln = (lc <= UNTREEIFY_THRESHOLD) ? untreeify(lo) : + (hc != 0) ? new TreeBin(lo) : t; + hn = (hc <= UNTREEIFY_THRESHOLD) ? untreeify(hi) : + (lc != 0) ? new TreeBin(hi) : t; + // 低位树的位置不变 + setTabAt(nextTab, i, ln); + // 高位树的位置是原位置加n + setTabAt(nextTab, i + n, hn); + // 标记该桶已迁移 + setTabAt(tab, i, fwd); + // advance为true，返回上面进行--i操作 + advance = true; + } + } + } + } + } +} +``` + +（1）新桶数组大小是旧桶数组的两倍； + +（2）迁移元素先从靠后的桶开始； + +（3）迁移完成的桶在里面放置一ForwardingNode类型的元素，标记该桶迁移完成； + +（4）迁移时根据hash&n是否等于0把桶中元素分化成两个链表或树； + +（5）低位链表（树）存储在原来的位置； + +（6）高们链表（树）存储在原来的位置加n的位置； + +（7）迁移元素时会锁住当前桶，也是分段锁的思想； + + + +### 删除元素 + +删除元素跟添加元素一样，都是先找到元素所在的桶，然后采用分段锁的思想锁住整个桶，再进行操作。 + +``` +public V remove(Object key) { + // 调用替换节点方法 + return replaceNode(key, null, null); +} + +final V replaceNode(Object key, V value, Object cv) { + // 计算hash + int hash = spread(key.hashCode()); + // 自旋 + for (Node[] tab = table;;) { + Node f; int n, i, fh; + if (tab == null || (n = tab.length) == 0 || + (f = tabAt(tab, i = (n - 1) & hash)) == null) + // 如果目标key所在的桶不存在，跳出循环返回null + break; + else if ((fh = f.hash) == MOVED) + // 如果正在扩容中，协助扩容 + tab = helpTransfer(tab, f); + else { + V oldVal = null; + // 标记是否处理过 + boolean validated = false; + synchronized (f) { + // 再次验证当前桶第一个元素是否被修改过 + if (tabAt(tab, i) == f) { + if (fh >= 0) { + // fh>=0表示是链表节点 + validated = true; + // 遍历链表寻找目标节点 + for (Node e = f, pred = null;;) { + K ek; + if (e.hash == hash && + ((ek = e.key) == key || + (ek != null && key.equals(ek)))) { + // 找到了目标节点 + V ev = e.val; + // 检查目标节点旧value是否等于cv + if (cv == null || cv == ev || + (ev != null && cv.equals(ev))) { + oldVal = ev; + if (value != null) + // 如果value不为空则替换旧值 + e.val = value; + else if (pred != null) + // 如果前置节点不为空 + // 删除当前节点 + pred.next = e.next; + else + // 如果前置节点为空 + // 说明是桶中第一个元素，删除之 + setTabAt(tab, i, e.next); + } + break; + } + pred = e; + // 遍历到链表尾部还没找到元素，跳出循环 + if ((e = e.next) == null) + break; + } + } + else if (f instanceof TreeBin) { + // 如果是树节点 + validated = true; + TreeBin t = (TreeBin)f; + TreeNode r, p; + // 遍历树找到了目标节点 + if ((r = t.root) != null && + (p = r.findTreeNode(hash, key, null)) != null) { + V pv = p.val; + // 检查目标节点旧value是否等于cv + if (cv == null || cv == pv || + (pv != null && cv.equals(pv))) { + oldVal = pv; + if (value != null) + // 如果value不为空则替换旧值 + p.val = value; + else if (t.removeTreeNode(p)) + // 如果value为空则删除元素 + // 如果删除后树的元素个数较少则退化成链表 + // t.removeTreeNode(p)这个方法返回true表示删除节点后树的元素个数较少 + setTabAt(tab, i, untreeify(t.first)); + } + } + } + } + } + // 如果处理过，不管有没有找到元素都返回 + if (validated) { + // 如果找到了元素，返回其旧值 + if (oldVal != null) { + // 如果要替换的值为空，元素个数减1 + if (value == null) + addCount(-1L, -1); + return oldVal; + } + break; + } + } + } + // 没找到元素返回空 + return null; +} +``` + +（1）计算hash； + +（2）如果所在的桶不存在，表示没有找到目标元素，返回； + +（3）如果正在扩容，则协助扩容完成后再进行删除操作； + +（4）如果是以链表形式存储的，则遍历整个链表查找元素，找到之后再删除； + +（5）如果是以树形式存储的，则遍历树查找元素，找到之后再删除； + +（6）如果是以树形式存储的，删除元素之后树较小，则退化成链表； + +（7）如果确实删除了元素，则整个map元素个数减1，并返回旧值； + +（8）如果没有删除元素，则返回null； + +### 获取元素 + +获取元素，根据目标key所在桶的第一个元素的不同采用不同的方式获取元素，关键点在于find()方法的重写。 + +``` +public V get(Object key) { + Node[] tab; Node e, p; int n, eh; K ek; + // 计算hash + int h = spread(key.hashCode()); + // 如果元素所在的桶存在且里面有元素 + if ((tab = table) != null && (n = tab.length) > 0 && + (e = tabAt(tab, (n - 1) & h)) != null) { + // 如果第一个元素就是要找的元素，直接返回 + if ((eh = e.hash) == h) { + if ((ek = e.key) == key || (ek != null && key.equals(ek))) + return e.val; + } + else if (eh < 0) + // hash小于0，说明是树或者正在扩容 + // 使用find寻找元素，find的寻找方式依据Node的不同子类有不同的实现方式 + 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; +} +``` + +（1）hash到元素所在的桶； + +（2）如果桶中第一个元素就是该找的元素，直接返回； + +（3）如果是树或者正在迁移元素，则调用各自Node子类的find()方法寻找元素； + +（4）如果是链表，遍历整个链表寻找元素； + +（5）获取元素没有加锁； + +### 获取元素个数 + +元素个数的存储也是采用分段的思想，获取元素个数时需要把所有段加起来。 + +``` +public int size() { + // 调用sumCount()计算元素个数 + long n = sumCount(); + return ((n < 0L) ? 0 : + (n > (long)Integer.MAX_VALUE) ? Integer.MAX_VALUE : + (int)n); +} + +final long sumCount() { + // 计算CounterCell所有段及baseCount的数量之和 + 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; +} +``` + +（1）元素的个数依据不同的线程存在在不同的段里；（见addCounter()分析） + +（2）计算CounterCell所有段及baseCount的数量之和； + +（3）获取元素个数没有加锁； + +## 总结 + +（1）ConcurrentHashMap是HashMap的线程安全版本； + +（2）ConcurrentHashMap采用（数组 + 链表 + 红黑树）的结构存储元素； + +（3）ConcurrentHashMap相比于同样线程安全的HashTable，效率要高很多； + +（4）ConcurrentHashMap采用的锁有 synchronized，CAS，自旋锁，分段锁，volatile等； + +（5）ConcurrentHashMap中没有threshold和loadFactor这两个字段，而是采用sizeCtl来控制； + +（6）sizeCtl = -1，表示正在进行初始化； + +（7）sizeCtl = 0，默认值，表示后续在真正初始化的时候使用默认容量； + +（8）sizeCtl > 0，在初始化之前存储的是传入的容量，在初始化或扩容后存储的是下一次的扩容门槛； + +（9）sizeCtl = (resizeStamp << 16) + (1 + nThreads)，表示正在进行扩容，高位存储扩容邮戳，低位存储扩容线程数加1； + +（10）更新操作时如果正在进行扩容，当前线程协助扩容； + +（11）更新操作会采用synchronized锁住当前桶的第一个元素，这是分段锁的思想； + +（12）整个扩容过程都是通过CAS控制sizeCtl这个字段来进行的，这很关键； + +（13）迁移完元素的桶会放置一个ForwardingNode节点，以标识该桶迁移完毕； + +（14）元素个数的存储也是采用的分段思想，类似于LongAdder的实现； + +（15）元素个数的更新会把不同的线程hash到不同的段上，减少资源争用； + +（16）元素个数的更新如果还是出现多个线程同时更新一个段，则会扩容段（CounterCell）； + +（17）获取元素个数是把所有的段（包括baseCount和CounterCell）相加起来得到的； + +（18）查询操作是不会加锁的，所以ConcurrentHashMap不是强一致性的； + +（19）ConcurrentHashMap中不能存储key或value为null的元素； + +## 彩蛋——值得学习的技术 + +ConcurrentHashMap中有哪些值得学习的技术呢？ + +我认为有以下几点： + +（1）CAS + 自旋，乐观锁的思想，减少线程上下文切换的时间； + +（2）分段锁的思想，减少同一把锁争用带来的低效问题； + +（3）CounterCell，分段存储元素个数，减少多线程同时更新一个字段带来的低效； + +（4）@sun.misc.Contended（CounterCell上的注解），避免伪共享；（p.s.伪共享我们后面也会讲的^^） + +（5）多线程协同进行扩容； + +（6）你又学到了哪些呢？ + +## 彩蛋——不能解决的问题 + +ConcurrentHashMap不能解决什么问题呢？ + +请看下面的例子： + +``` +private static final Map map = new ConcurrentHashMap<>(); + +public void unsafeUpdate(Integer key, Integer value) { + Integer oldValue = map.get(key); + if (oldValue == null) { + map.put(key, value); + } +} +``` + +这里如果有多个线程同时调用unsafeUpdate()这个方法，ConcurrentHashMap还能保证线程安全吗？ + +答案是不能。因为get()之后if之前可能有其它线程已经put()了这个元素，这时候再put()就把那个线程put()的元素覆盖了。 + +那怎么修改呢？ + +答案也很简单，使用putIfAbsent()方法，它会保证元素不存在时才插入元素，如下： + +``` +public void safeUpdate(Integer key, Integer value) { + map.putIfAbsent(key, value); +} +``` + +那么，如果上面oldValue不是跟null比较，而是跟一个特定的值比如1进行比较怎么办？也就是下面这样： + +``` +public void unsafeUpdate(Integer key, Integer value) { + Integer oldValue = map.get(key); + if (oldValue == 1) { + map.put(key, value); + } +} +``` + +这样的话就没办法使用putIfAbsent()方法了。 + +其实，ConcurrentHashMap还提供了另一个方法叫replace(K key, V oldValue, V newValue)可以解决这个问题。 + +replace(K key, V oldValue, V newValue)这个方法可不能乱用，如果传入的newValue是null，则会删除元素。 + +``` +public void safeUpdate(Integer key, Integer value) { + map.replace(key, 1, value); +} +``` + +那么，如果if之后不是简单的put()操作，而是还有其它业务操作，之后才是put()，比如下面这样，这该怎么办呢？ + +``` +public void unsafeUpdate(Integer key, Integer value) { + Integer oldValue = map.get(key); + if (oldValue == 1) { + System.out.println(System.currentTimeMillis()); + /** + * 其它业务操作 + */ + System.out.println(System.currentTimeMillis()); + + map.put(key, value); + } +} +``` + +这时候就没办法使用ConcurrentHashMap提供的方法了，只能业务自己来保证线程安全了，比如下面这样： + +``` +public void safeUpdate(Integer key, Integer value) { + synchronized (map) { + Integer oldValue = map.get(key); + if (oldValue == null) { + System.out.println(System.currentTimeMillis()); + /** + * 其它业务操作 + */ + System.out.println(System.currentTimeMillis()); + + map.put(key, value); + } + } +} +``` + +这样虽然不太友好，但是最起码能保证业务逻辑是正确的。 + +当然，这里使用ConcurrentHashMap的意义也就不大了，可以换成普通的HashMap了。 + +上面只是举一个简单的例子，我们不能听说ConcurrentHashMap是线程安全的，就认为它无论什么情况下都是线程安全的，还是那句话尽信书不如无书。 + +这也正是我们读源码的目的之一，了解其本质，才能在我们的实际工作中少挖坑，不论是挖给别人还是挖给自己^^ \ No newline at end of file diff --git a/week_04/51/ConcurrentLinkedQueue_51.md b/week_04/51/ConcurrentLinkedQueue_51.md new file mode 100644 index 0000000..d6c6c67 --- /dev/null +++ b/week_04/51/ConcurrentLinkedQueue_51.md @@ -0,0 +1,217 @@ +## 问题 + +（1）ConcurrentLinkedQueue是阻塞队列吗？ + +（2）ConcurrentLinkedQueue如何保证并发安全？ + +（3）ConcurrentLinkedQueue能用于线程池吗？ + +## 简介 + +![qrcode](https://gitee.com/alan-tang-tt/yuan/raw/master/%E6%AD%BB%E7%A3%95%20java%E9%9B%86%E5%90%88%E7%B3%BB%E5%88%97/resource/ConcurrentLinkedQueue.png) + +ConcurrentLinkedQueue只实现了Queue接口，并没有实现BlockingQueue接口，所以它不是阻塞队列，也不能用于线程池中，但是它是线程安全的，可用于多线程环境中。 + +那么，它的线程安全又是如何实现的呢？让我们一起来瞧一瞧。 + +## 源码分析 + +### 主要属性 + +``` +// 链表头节点 +private transient volatile Node head; +// 链表尾节点 +private transient volatile Node tail; +``` + +就这两个主要属性，一个头节点，一个尾节点。 + +### 主要内部类 + +``` +private static class Node { + volatile E item; + volatile Node next; +} +``` + +典型的单链表结构，非常纯粹。 + +### 主要构造方法 + +``` +public ConcurrentLinkedQueue() { + // 初始化头尾节点 + head = tail = new Node(null); +} + +public ConcurrentLinkedQueue(Collection c) { + Node h = null, t = null; + // 遍历c，并把它元素全部添加到单链表中 + for (E e : c) { + checkNotNull(e); + Node newNode = new Node(e); + if (h == null) + h = t = newNode; + else { + t.lazySetNext(newNode); + t = newNode; + } + } + if (h == null) + h = t = new Node(null); + head = h; + tail = t; +} +``` + +这两个构造方法也很简单，可以看到这是一个无界的单链表实现的队列。 + +### 入队 + +因为它不是阻塞队列，所以只有两个入队的方法，add(e)和offer(e)。 + +因为是无界队列，所以add(e)方法也不用抛出异常了。 + +``` +public boolean add(E e) { + return offer(e); +} + +public boolean offer(E e) { + // 不能添加空元素 + checkNotNull(e); + // 新节点 + final Node newNode = new Node(e); + + // 入队到链表尾 + for (Node t = tail, p = t;;) { + Node q = p.next; + // 如果没有next，说明到链表尾部了，就入队 + if (q == null) { + // CAS更新p的next为新节点 + // 如果成功了，就返回true + // 如果不成功就重新取next重新尝试 + if (p.casNext(null, newNode)) { + // 如果p不等于t，说明有其它线程先一步更新tail + // 也就不会走到q==null这个分支了 + // p取到的可能是t后面的值 + // 把tail原子更新为新节点 + if (p != t) // hop two nodes at a time + casTail(t, newNode); // Failure is OK. + // 返回入队成功 + return true; + } + } + else if (p == q) + // 如果p的next等于p，说明p已经被删除了（已经出队了） + // 重新设置p的值 + p = (t != (t = tail)) ? t : head; + else + // t后面还有值，重新设置p的值 + p = (p != t && t != (t = tail)) ? t : q; + } +} +``` + +入队整个流程还是比较清晰的，这里有个前提是出队时会把出队的那个节点的next设置为节点本身。 + +（1）定位到链表尾部，尝试把新节点放到后面； + +（2）如果尾部变化了，则重新获取尾部，再重试； + +### 出队 + +因为它不是阻塞队列，所以只有两个出队的方法，remove()和poll()。 + +``` +public E remove() { + E x = poll(); + if (x != null) + return x; + else + throw new NoSuchElementException(); +} + +public E poll() { + restartFromHead: + for (;;) { + // 尝试弹出链表的头节点 + for (Node h = head, p = h, q;;) { + E item = p.item; + // 如果节点的值不为空，并且将其更新为null成功了 + if (item != null && p.casItem(item, null)) { + // 如果头节点变了，则不会走到这个分支 + // 会先走下面的分支拿到新的头节点 + // 这时候p就不等于h了，就更新头节点 + // 在updateHead()中会把head更新为新节点 + // 并让head的next指向其自己 + if (p != h) // hop two nodes at a time + updateHead(h, ((q = p.next) != null) ? q : p); + // 上面的casItem()成功，就可以返回出队的元素了 + return item; + } + // 下面三个分支说明头节点变了 + // 且p的item肯定为null + else if ((q = p.next) == null) { + // 如果p的next为空，说明队列中没有元素了 + // 更新h为p，也就是空元素的节点 + updateHead(h, p); + // 返回null + return null; + } + else if (p == q) + // 如果p等于p的next，说明p已经出队了，重试 + continue restartFromHead; + else + // 将p设置为p的next + p = q; + } + } +} +// 更新头节点的方法 +final void updateHead(Node h, Node p) { + // 原子更新h为p成功后，延迟更新h的next为它自己 + // 这里用延迟更新是安全的，因为head节点已经变了 + // 只要入队出队的时候检查head有没有变化就行了，跟它的next关系不大 + if (h != p && casHead(h, p)) + h.lazySetNext(h); +} +``` + +出队的整个逻辑也是比较清晰的： + +（1）定位到头节点，尝试更新其值为null； + +（2）如果成功了，就成功出队； + +（3）如果失败或者头节点变化了，就重新寻找头节点，并重试； + +（4）整个出队过程没有一点阻塞相关的代码，所以出队的时候不会阻塞线程，没找到元素就返回null； + +## 总结 + +（1）ConcurrentLinkedQueue不是阻塞队列； + +（2）ConcurrentLinkedQueue不能用在线程池中； + +（3）ConcurrentLinkedQueue使用（CAS+自旋）更新头尾节点控制出队入队操作； + +## 彩蛋 + +ConcurrentLinkedQueue与LinkedBlockingQueue对比？ + +（1）两者都是线程安全的队列； + +（2）两者都可以实现取元素时队列为空直接返回null，后者的poll()方法可以实现此功能； + +（3）前者全程无锁，后者全部都是使用重入锁控制的； + +（4）前者效率较高，后者效率较低； + +（5）前者无法实现如果队列为空等待元素到来的操作； + +（6）前者是非阻塞队列，后者是阻塞队列； + +（7）前者无法用在线程池中，后者可以； \ No newline at end of file diff --git a/week_04/51/CopyOnWriteArrayList_51.md b/week_04/51/CopyOnWriteArrayList_51.md new file mode 100644 index 0000000..2fb7be0 --- /dev/null +++ b/week_04/51/CopyOnWriteArrayList_51.md @@ -0,0 +1,355 @@ +## 简介 + +CopyOnWriteArrayList是ArrayList的线程安全版本，内部也是通过数组实现，每次对数组的修改都完全拷贝一份新的数组来修改，修改完了再替换掉老数组，这样保证了只阻塞写操作，不阻塞读操作，实现读写分离。 + +## 继承体系 + +![CopyOnWriteArrayList](https://gitee.com/alan-tang-tt/yuan/raw/master/%E6%AD%BB%E7%A3%95%20java%E9%9B%86%E5%90%88%E7%B3%BB%E5%88%97/resource/CopyOnWriteArrayList.png) + +CopyOnWriteArrayList实现了List, RandomAccess, Cloneable, java.io.Serializable等接口。 + +CopyOnWriteArrayList实现了List，提供了基础的添加、删除、遍历等操作。 + +CopyOnWriteArrayList实现了RandomAccess，提供了随机访问的能力。 + +CopyOnWriteArrayList实现了Cloneable，可以被克隆。 + +CopyOnWriteArrayList实现了Serializable，可以被序列化。 + +## 源码解析 + +### 属性 + +``` +/** 用于修改时加锁 */ +final transient ReentrantLock lock = new ReentrantLock(); + +/** 真正存储元素的地方，只能通过getArray()/setArray()访问 */ +private transient volatile Object[] array; +``` + +（1）lock + +用于修改时加锁，使用transient修饰表示不自动序列化。 + +（2）array + +真正存储元素的地方，使用transient修饰表示不自动序列化，使用volatile修饰表示一个线程对这个字段的修改另外一个线程立即可见。 + +*问题：为啥没有size字段？且听后续分解。* + +### CopyOnWriteArrayList()构造方法 + +创建空数组。 + +``` +public CopyOnWriteArrayList() { + // 所有对array的操作都是通过setArray()和getArray()进行 + setArray(new Object[0]); +} + +final void setArray(Object[] a) { + array = a; +} +``` + +### CopyOnWriteArrayList(Collection c)构造方法 + +如果c是CopyOnWriteArrayList类型，直接把它的数组赋值给当前list的数组，注意这里是浅拷贝，两个集合共用同一个数组。 + +如果c不是CopyOnWriteArrayList类型，则进行拷贝把c的元素全部拷贝到当前list的数组中。 + +``` +public CopyOnWriteArrayList(Collection c) { + Object[] elements; + if (c.getClass() == CopyOnWriteArrayList.class) + // 如果c也是CopyOnWriteArrayList类型 + // 那么直接把它的数组拿过来使用 + elements = ((CopyOnWriteArrayList)c).getArray(); + else { + // 否则调用其toArray()方法将集合元素转化为数组 + elements = c.toArray(); + // 这里c.toArray()返回的不一定是Object[]类型 + // 详细原因见ArrayList里面的分析 + if (elements.getClass() != Object[].class) + elements = Arrays.copyOf(elements, elements.length, Object[].class); + } + setArray(elements); +} +``` + +### CopyOnWriteArrayList(E[] toCopyIn)构造方法 + +把toCopyIn的元素拷贝给当前list的数组。 + +``` +public CopyOnWriteArrayList(E[] toCopyIn) { + setArray(Arrays.copyOf(toCopyIn, toCopyIn.length, Object[].class)); +} +``` + +### add(E e)方法 + +添加一个元素到末尾。 + +``` +public boolean add(E e) { + final ReentrantLock lock = this.lock; + // 加锁 + lock.lock(); + try { + // 获取旧数组 + Object[] elements = getArray(); + int len = elements.length; + // 将旧数组元素拷贝到新数组中 + // 新数组大小是旧数组大小加1 + Object[] newElements = Arrays.copyOf(elements, len + 1); + // 将元素放在最后一位 + newElements[len] = e; + setArray(newElements); + return true; + } finally { + // 释放锁 + lock.unlock(); + } +} +``` + +（1）加锁； + +（2）获取元素数组； + +（3）新建一个数组，大小为原数组长度加1，并把原数组元素拷贝到新数组； + +（4）把新添加的元素放到新数组的末尾； + +（5）把新数组赋值给当前对象的array属性，覆盖原数组； + +（6）解锁； + +### add(int index, E element)方法 + +添加一个元素在指定索引处。 + +``` +public void add(int index, E element) { + final ReentrantLock lock = this.lock; + // 加锁 + lock.lock(); + try { + // 获取旧数组 + Object[] elements = getArray(); + int len = elements.length; + // 检查是否越界, 可以等于len + if (index > len || index < 0) + throw new IndexOutOfBoundsException("Index: "+index+ + ", Size: "+len); + Object[] newElements; + int numMoved = len - index; + if (numMoved == 0) + // 如果插入的位置是最后一位 + // 那么拷贝一个n+1的数组, 其前n个元素与旧数组一致 + newElements = Arrays.copyOf(elements, len + 1); + else { + // 如果插入的位置不是最后一位 + // 那么新建一个n+1的数组 + newElements = new Object[len + 1]; + // 拷贝旧数组前index的元素到新数组中 + System.arraycopy(elements, 0, newElements, 0, index); + // 将index及其之后的元素往后挪一位拷贝到新数组中 + // 这样正好index位置是空出来的 + System.arraycopy(elements, index, newElements, index + 1, + numMoved); + } + // 将元素放置在index处 + newElements[index] = element; + setArray(newElements); + } finally { + // 释放锁 + lock.unlock(); + } +} +``` + +（1）加锁； + +（2）检查索引是否合法，如果不合法抛出IndexOutOfBoundsException异常，注意这里index等于len也是合法的； + +（3）如果索引等于数组长度（也就是数组最后一位再加1），那就拷贝一个len+1的数组； + +（4）如果索引不等于数组长度，那就新建一个len+1的数组，并按索引位置分成两部分，索引之前（不包含）的部分拷贝到新数组索引之前（不包含）的部分，索引之后（包含）的位置拷贝到新数组索引之后（不包含）的位置，索引所在位置留空； + +（5）把索引位置赋值为待添加的元素； + +（6）把新数组赋值给当前对象的array属性，覆盖原数组； + +（7）解锁； + +### addIfAbsent(E e)方法 + +添加一个元素如果这个元素不存在于集合中。 + +``` +public boolean addIfAbsent(E e) { + // 获取元素数组, 取名为快照 + Object[] snapshot = getArray(); + // 检查如果元素不存在,直接返回false + // 如果存在再调用addIfAbsent()方法添加元素 + return indexOf(e, snapshot, 0, snapshot.length) >= 0 ? false : + addIfAbsent(e, snapshot); +} + +private boolean addIfAbsent(E e, Object[] snapshot) { + final ReentrantLock lock = this.lock; + // 加锁 + lock.lock(); + try { + // 重新获取旧数组 + Object[] current = getArray(); + int len = current.length; + // 如果快照与刚获取的数组不一致 + // 说明有修改 + if (snapshot != current) { + // 重新检查元素是否在刚获取的数组里 + int common = Math.min(snapshot.length, len); + for (int i = 0; i < common; i++) + // 到这个方法里面了, 说明元素不在快照里面 + if (current[i] != snapshot[i] && eq(e, current[i])) + return false; + if (indexOf(e, current, common, len) >= 0) + return false; + } + // 拷贝一份n+1的数组 + Object[] newElements = Arrays.copyOf(current, len + 1); + // 将元素放在最后一位 + newElements[len] = e; + setArray(newElements); + return true; + } finally { + // 释放锁 + lock.unlock(); + } +} +``` + +（1）检查这个元素是否存在于数组快照中； + +（2）如果存在直接返回false，如果不存在调用addIfAbsent(E e, Object[] snapshot)处理; + +（3）加锁； + +（4）如果当前数组不等于传入的快照，说明有修改，检查待添加的元素是否存在于当前数组中，如果存在直接返回false; + +（5）拷贝一个新数组，长度等于原数组长度加1，并把原数组元素拷贝到新数组中； + +（6）把新元素添加到数组最后一位； + +（7）把新数组赋值给当前对象的array属性，覆盖原数组； + +（8）解锁； + +### get(int index) + +获取指定索引的元素，支持随机访问，时间复杂度为O(1)。 + +``` +public E get(int index) { + // 获取元素不需要加锁 + // 直接返回index位置的元素 + // 这里是没有做越界检查的, 因为数组本身会做越界检查 + return get(getArray(), index); +} + +final Object[] getArray() { + return array; +} + +private E get(Object[] a, int index) { + return (E) a[index]; +} +``` + +（1）获取元素数组； + +（2）返回数组指定索引位置的元素； + +### remove(int index)方法 + +删除指定索引位置的元素。 + +``` +public E remove(int index) { + final ReentrantLock lock = this.lock; + // 加锁 + lock.lock(); + try { + // 获取旧数组 + Object[] elements = getArray(); + int len = elements.length; + E oldValue = get(elements, index); + int numMoved = len - index - 1; + if (numMoved == 0) + // 如果移除的是最后一位 + // 那么直接拷贝一份n-1的新数组, 最后一位就自动删除了 + setArray(Arrays.copyOf(elements, len - 1)); + else { + // 如果移除的不是最后一位 + // 那么新建一个n-1的新数组 + Object[] newElements = new Object[len - 1]; + // 将前index的元素拷贝到新数组中 + System.arraycopy(elements, 0, newElements, 0, index); + // 将index后面(不包含)的元素往前挪一位 + // 这样正好把index位置覆盖掉了, 相当于删除了 + System.arraycopy(elements, index + 1, newElements, index, + numMoved); + setArray(newElements); + } + return oldValue; + } finally { + // 释放锁 + lock.unlock(); + } +} +``` + +（1）加锁； + +（2）获取指定索引位置元素的旧值； + +（3）如果移除的是最后一位元素，则把原数组的前len-1个元素拷贝到新数组中，并把新数组赋值给当前对象的数组属性； + +（4）如果移除的不是最后一位元素，则新建一个len-1长度的数组，并把原数组除了指定索引位置的元素全部拷贝到新数组中，并把新数组赋值给当前对象的数组属性； + +（5）解锁并返回旧值； + +### size()方法 + +返回数组的长度。 + +``` +public int size() { + // 获取元素个数不需要加锁 + // 直接返回数组的长度 + return getArray().length; +} +``` + +## 总结 + +（1）CopyOnWriteArrayList使用ReentrantLock重入锁加锁，保证线程安全； + +（2）CopyOnWriteArrayList的写操作都要先拷贝一份新数组，在新数组中做修改，修改完了再用新数组替换老数组，所以空间复杂度是O(n)，性能比较低下； + +（3）CopyOnWriteArrayList的读操作支持随机访问，时间复杂度为O(1)； + +（4）CopyOnWriteArrayList采用读写分离的思想，读操作不加锁，写操作加锁，且写操作占用较大内存空间，所以适用于读多写少的场合； + +（5）CopyOnWriteArrayList只保证最终一致性，不保证实时一致性； + +## 彩蛋 + +*为什么CopyOnWriteArrayList没有size属性？* + +因为每次修改都是拷贝一份正好可以存储目标个数元素的数组，所以不需要size属性了，数组的长度就是集合的大小，而不像ArrayList数组的长度实际是要大于集合的大小的。 + +比如，add(E e)操作，先拷贝一份n+1个元素的数组，再把新元素放到新数组的最后一位，这时新数组的长度为len+1了，也就是集合的size了。 \ No newline at end of file diff --git "a/week_04/51/Week4_51\346\200\273\347\273\223.md" "b/week_04/51/Week4_51\346\200\273\347\273\223.md" new file mode 100644 index 0000000..4349443 --- /dev/null +++ "b/week_04/51/Week4_51\346\200\273\347\273\223.md" @@ -0,0 +1,75 @@ +## CopyOnWriteArrayList + +（1）CopyOnWriteArrayList使用ReentrantLock重入锁加锁，保证线程安全； + +（2）CopyOnWriteArrayList的写操作都要先拷贝一份新数组，在新数组中做修改，修改完了再用新数组替换老数组，所以空间复杂度是O(n)，性能比较低下； + +（3）CopyOnWriteArrayList的读操作支持随机访问，时间复杂度为O(1)； + +（4）CopyOnWriteArrayList采用读写分离的思想，读操作不加锁，写操作加锁，且写操作占用较大内存空间，所以适用于读多写少的场合； + +（5）CopyOnWriteArrayList只保证最终一致性，不保证实时一致性； + +对于一些读多写少的数据，这种做法的确很不错，例如配置、黑名单、物流地址等变化非常少的数据，这是一种无锁的实现。可以帮我们实现程序更高的并发。 + +这种实现只是保证数据的最终一致性，在添加到拷贝数据而还没进行替换的时候，读到的仍然是旧数据。如果对象比较大，频繁地进行替换会消耗内存，从而引发Java的GC问题，这个时候，我们应该考虑其他的容器，例如ConcurrentHashMap。 + +## ConcurrentHashMap + +（1）ConcurrentHashMap是HashMap的线程安全版本； + +（2）ConcurrentHashMap采用（数组 + 链表 + 红黑树）的结构存储元素； + +（3）ConcurrentHashMap相比于同样线程安全的HashTable，效率要高很多； + +（4）ConcurrentHashMap采用的锁有 synchronized，CAS，自旋锁，分段锁，volatile等； + +（5）ConcurrentHashMap中没有threshold和loadFactor这两个字段，而是采用sizeCtl来控制； + +（6）sizeCtl = -1，表示正在进行初始化； + +（7）sizeCtl = 0，默认值，表示后续在真正初始化的时候使用默认容量； + +（8）sizeCtl > 0，在初始化之前存储的是传入的容量，在初始化或扩容后存储的是下一次的扩容门槛； + +（9）sizeCtl = (resizeStamp << 16) + (1 + nThreads)，表示正在进行扩容，高位存储扩容邮戳，低位存储扩容线程数加1； + +（10）更新操作时如果正在进行扩容，当前线程协助扩容； + +（11）更新操作会采用synchronized锁住当前桶的第一个元素，这是分段锁的思想； + +（12）整个扩容过程都是通过CAS控制sizeCtl这个字段来进行的，这很关键； + +（13）迁移完元素的桶会放置一个ForwardingNode节点，以标识该桶迁移完毕； + +（14）元素个数的存储也是采用的分段思想，类似于LongAdder的实现； + +（15）元素个数的更新会把不同的线程hash到不同的段上，减少资源争用； + +（16）元素个数的更新如果还是出现多个线程同时更新一个段，则会扩容段（CounterCell）； + +（17）获取元素个数是把所有的段（包括baseCount和CounterCell）相加起来得到的； + +（18）查询操作是不会加锁的，所以ConcurrentHashMap不是强一致性的； + +（19）ConcurrentHashMap中不能存储key或value为null的元素； + + + +## ConcurrentLinkedQueue + +（1）ConcurrentLinkedQueue不是阻塞队列； + +（2）ConcurrentLinkedQueue不能用在线程池中； + +（3）ConcurrentLinkedQueue使用（CAS+自旋）更新头尾节点控制出队入队操作； + +ConcurrentLinkedQueue 的非阻塞算法实现可概括为下面 5 点： + +使用 CAS 原子指令来处理对数据的并发访问，这是非阻塞算法得以实现的基础。 +head/tail 并非总是指向队列的头 / 尾节点，也就是说允许队列处于不一致状态。 这个特性把入队 / 出队时，原本需要一起原子化执行的两个步骤分离开来，从而缩小了入队 / 出队时需要原子化更新值的范围到唯一变量。这是非阻塞算法得以实现的关键。 +由于队列有时会处于不一致状态。为此，ConcurrentLinkedQueue 使用三个不变式来维护非阻塞算法的正确性。 +以批处理方式来更新 head/tail，从整体上减少入队 / 出队操作的开销。 +为了有利于垃圾收集，队列使用特有的 head 更新机制；为了确保从已删除节点向后遍历，可到达所有的非删除节点，队列使用了特有的向后推进策略。 + +原文链接：https://blog.csdn.net/qq_38293564/article/details/80798310 \ No newline at end of file -- Gitee