diff --git a/week_04/44/ArrayBlockingQueue.md b/week_04/44/ArrayBlockingQueue.md new file mode 100644 index 0000000000000000000000000000000000000000..0545eb77520d00c295bbf841d5c2cfb79214843c --- /dev/null +++ b/week_04/44/ArrayBlockingQueue.md @@ -0,0 +1,277 @@ +# ArrayBlockingQueue + +## 参考 + +彤哥读源码 + +## 简介 + +队列，是一种线性表，它的特点是先进先出，又叫FIFO，就像我们平常排队一样，先到先得，即先进入队列的人先出队。 + +ArrayBlockingQueue是java并发包下一个以数组实现的阻塞队列，它是线程安全的。 + +## 源码分析 + +### 属性 + +```java +// 使用数组存储元素 +final Object[] items; + +// 取元素的指针 +int takeIndex; + +// 放元素的指针 +int putIndex; + +// 元素数量 +int count; + +// 保证并发访问的锁 +final ReentrantLock lock; + +// 非空条件 +private final Condition notEmpty; + +// 非满条件 +private final Condition notFull; +``` + +通过属性我们可以得出以下几个重要信息： + +（1）利用数组存储元素； + +（2）通过放指针和取指针来标记下一次操作的位置； + +（3）利用重入锁来保证并发安全； + +### 主要构造方法 + +```java +public ArrayBlockingQueue(int capacity) { + this(capacity, false); +} + +public ArrayBlockingQueue(int capacity, boolean fair) { + if (capacity <= 0) + throw new IllegalArgumentException(); + // 初始化数组 + this.items = new Object[capacity]; + // 创建重入锁及两个条件 + lock = new ReentrantLock(fair); + notEmpty = lock.newCondition(); + notFull = lock.newCondition(); +} +``` + +通过构造方法我们可以得出以下两个结论： + +（1）ArrayBlockingQueue初始化时必须传入容量，也就是数组的大小； + +（2）可以通过构造方法控制重入锁的类型是公平锁还是非公平锁； + +### 入队 + +入队有四个方法，它们分别是add(E e)、offer(E e)、put(E e)、offer(E e, long timeout, TimeUnit unit)，它们有什么区别呢？ + +```java +public boolean add(E e) { + // 调用父类的add(e)方法 + return super.add(e); +} + +// super.add(e) +public boolean add(E e) { + // 调用offer(e)如果成功返回true，如果失败抛出异常 + if (offer(e)) + return true; + else + throw new IllegalStateException("Queue full"); +} + +public boolean offer(E e) { + // 元素不可为空 + checkNotNull(e); + final ReentrantLock lock = this.lock; + // 加锁 + lock.lock(); + try { + if (count == items.length) + // 如果数组满了就返回false + return false; + else { + // 如果数组没满就调用入队方法并返回true + enqueue(e); + return true; + } + } finally { + // 解锁 + lock.unlock(); + } +} + +public void put(E e) throws InterruptedException { + checkNotNull(e); + final ReentrantLock lock = this.lock; + // 加锁，如果线程中断了抛出异常 + lock.lockInterruptibly(); + try { + // 如果数组满了，使用notFull等待 + // notFull等待的意思是说现在队列满了 + // 只有取走一个元素后，队列才不满 + // 然后唤醒notFull，然后继续现在的逻辑 + // 这里之所以使用while而不是if + // 是因为有可能多个线程阻塞在lock上 + // 即使唤醒了可能其它线程先一步修改了队列又变成满的了 + // 这时候需要再次等待 + while (count == items.length) + notFull.await(); + // 入队 + enqueue(e); + } finally { + // 解锁 + lock.unlock(); + } +} + +public boolean offer(E e, long timeout, TimeUnit unit) + throws InterruptedException { + checkNotNull(e); + long nanos = unit.toNanos(timeout); + final ReentrantLock lock = this.lock; + // 加锁 + lock.lockInterruptibly(); + try { + // 如果数组满了，就阻塞nanos纳秒 + // 如果唤醒这个线程时依然没有空间且时间到了就返回false + while (count == items.length) { + if (nanos <= 0) + return false; + nanos = notFull.awaitNanos(nanos); + } + // 入队 + enqueue(e); + return true; + } finally { + // 解锁 + lock.unlock(); + } +} + +private void enqueue(E x) { + final Object[] items = this.items; + // 把元素直接放在放指针的位置上 + items[putIndex] = x; + // 如果放指针到数组尽头了，就返回头部 + if (++putIndex == items.length) + putIndex = 0; + // 数量加1 + count++; + // 唤醒notEmpty，因为入队了一个元素，所以肯定不为空了 + notEmpty.signal(); +} +``` + +（1）add(e)时如果队列满了则抛出异常； + +（2）offer(e)时如果队列满了则返回false； + +（3）put(e)时如果队列满了则使用notFull等待； + +（4）offer(e, timeout, unit)时如果队列满了则等待一段时间后如果队列依然满就返回false； + +（5）利用放指针循环使用数组来存储元素； + +### 出队 + +出队有四个方法，它们分别是remove()、poll()、take()、poll(long timeout, TimeUnit unit)，它们有什么区别呢？ + +```java +public E remove() { + // 调用poll()方法出队 + E x = poll(); + if (x != null) + // 如果有元素出队就返回这个元素 + return x; + else + // 如果没有元素出队就抛出异常 + throw new NoSuchElementException(); +} + +public E poll() { + final ReentrantLock lock = this.lock; + // 加锁 + lock.lock(); + try { + // 如果队列没有元素则返回null，否则出队 + return (count == 0) ? null : dequeue(); + } finally { + lock.unlock(); + } +} + +public E take() throws InterruptedException { + final ReentrantLock lock = this.lock; + // 加锁 + lock.lockInterruptibly(); + try { + // 如果队列无元素，则阻塞等待在条件notEmpty上 + while (count == 0) + notEmpty.await(); + // 有元素了再出队 + return dequeue(); + } finally { + // 解锁 + lock.unlock(); + } +} + +public E poll(long timeout, TimeUnit unit) throws InterruptedException { + long nanos = unit.toNanos(timeout); + final ReentrantLock lock = this.lock; + // 加锁 + lock.lockInterruptibly(); + try { + // 如果队列无元素，则阻塞等待nanos纳秒 + // 如果下一次这个线程获得了锁但队列依然无元素且已超时就返回null + while (count == 0) { + if (nanos <= 0) + return null; + nanos = notEmpty.awaitNanos(nanos); + } + return dequeue(); + } finally { + lock.unlock(); + } +} + +private E dequeue() { + final Object[] items = this.items; + @SuppressWarnings("unchecked") + // 取取指针位置的元素 + E x = (E) items[takeIndex]; + // 把取指针位置设为null + items[takeIndex] = null; + // 取指针前移，如果数组到头了就返回数组前端循环利用 + if (++takeIndex == items.length) + takeIndex = 0; + // 元素数量减1 + count--; + if (itrs != null) + itrs.elementDequeued(); + // 唤醒notFull条件 + notFull.signal(); + return x; +} +``` + +（1）remove()时如果队列为空则抛出异常； + +（2）poll()时如果队列为空则返回null； + +（3）take()时如果队列为空则阻塞等待在条件notEmpty上； + +（4）poll(timeout, unit)时如果队列为空则阻塞等待一段时间后如果还为空就返回null； + +（5）利用取指针循环从数组中取元素 + diff --git a/week_04/44/ConcurrentHashMap.md b/week_04/44/ConcurrentHashMap.md new file mode 100644 index 0000000000000000000000000000000000000000..91f09cde4660927ee1bffc28428d42cde19e2d5c --- /dev/null +++ b/week_04/44/ConcurrentHashMap.md @@ -0,0 +1,872 @@ +# ConcurrentHashMap + +## 参考 + +彤哥读源码 + +《Java并发编程的艺术》 + +## 简介 + +​ 在并发编程中使用HashMap可能导致程序死循环。而使用线程安全的HashTable效率又非常低下，基于以上两个原因，便有了ConcurrentHashMap的登场机会。 +​ （1）线程不安全的HashMap +​ 在多线程环境下，使用HashMap进行put操作会引起死循环，导致CPU利用率接近100%，所以在并发情况下不能使用HashMap。 HashMap在并发执行put操作时会引起死循环，是因为多线程会导致HashMap的Entry链表形成环形数据结构，一旦形成环形数据结构，Entry的next节点永远不为空，就会产生死循环获取Entry。 +​ （2）效率低下的HashTable +​ HashTable容器使用synchronized来保证线程安全，但在线程竞争激烈的情况下HashTable的效率非常低下。因为当一个线程访问HashTable的同步方法，其他线程也访问HashTable的同步方法时，会进入阻塞或轮询状态。如线程1使用put进行元素添加，线程2不但不能使用put方法添加元素，也不能使用get方法来获取元素，所以竞争越激烈效率越低。 +​ （3）ConcurrentHashMap的锁分段技术可有效提升并发访问率 +​ HashTable容器在竞争激烈的并发环境下表现出效率低下的原因是所有访问HashTable的线程都必须竞争同一把锁，假如容器里有多把锁，每一把锁用于锁容器其中一部分数据，那么当多线程访问容器里不同数据段的数据时，线程间就不会存在锁竞争，从而可以有效提高并发访问效率，这就是ConcurrentHashMap所使用的锁分段技术。首先将数据分成一段一段地存储，然后给每一段数据配一把锁，当一个线程占用锁访问其中一个段数据的时候，其他段的数据也能被其他线程访问。 + +## 源码分析 + +### 自定义类 + +#### Node + +```java +static class Node implements Map.Entry { + final int hash; + final K key; + volatile V val; + volatile Node next; + + Node(int hash, K key, V val, Node next) { + this.hash = hash; + this.key = key; + this.val = val; + this.next = next; + } + + Node find(int h, Object k) +} +``` + +#### TreeNode + +```java +static final class TreeNode extends Node { + TreeNode parent; // red-black tree links + TreeNode left; + TreeNode right; + TreeNode prev; // needed to unlink next upon deletion + boolean red; + + TreeNode(int hash, K key, V val, Node next, TreeNode parent) { + super(hash, key, val, next); + this.parent = parent; + } + + Node find(int h, Object k) { + return findTreeNode(h, k, null); + } + + final TreeNode findTreeNode(int h, Object k, Class kc) +} +``` + +TreeBin + +```java +static final class TreeBin extends Node { + TreeNode root; + volatile TreeNode first; + volatile Thread waiter; + volatile int lockState; + // values for lockState + static final int WRITER = 1; // set while holding write lock + static final int WAITER = 2; // set when waiting for write lock + static final int READER = 4; // increment value for setting read lock + + TreeBin(TreeNode b) + private final void lockRoot() + final TreeNode putTreeVal(int h, K k, V v) + final boolean removeTreeNode(TreeNode p) + static TreeNode rotateLeft(TreeNode root, TreeNode p) + static TreeNode rotateRight(TreeNode root, TreeNode p) + static TreeNode balanceInsertion(TreeNode root, TreeNode x) + static TreeNode balanceDeletion(TreeNode root, TreeNode x) + static boolean checkInvariants(TreeNode t) +} +``` + +### 属性 + +```java +transient volatile Node[] table; + +private transient volatile Node[] nextTable; + +private transient volatile long baseCount; + +private transient volatile int sizeCtl; + +private transient volatile int transferIndex; + +private transient volatile int cellsBusy; + +private transient volatile CounterCell[] counterCells; + +private transient KeySetView keySet; +private transient ValuesView values; +private transient EntrySetView entrySet; +``` + +### 构造方法 + +```java +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，初始化或扩容完成后下一次的扩容门槛 + +### 添加元素 + +```java +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差。 + +### 扩容 + +#### 初始化桶数组 + +```java +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，并判断是否达到扩容门槛，达到了则进行扩容或协助扩容。 + +```java +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）其它线程添加元素后如果发现存在扩容，也会加入的扩容行列中来； + +#### 协助扩容 + +线程添加元素时发现正在扩容且当前元素所在的桶元素已经迁移完成了，则协助迁移其它桶的元素。 + +```java +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; +} +``` + +当前桶元素迁移完成了才去协助迁移其它桶元素； + +#### 迁移元素 + +扩容时容量变为两倍，并把部分元素迁移到其它桶中。 + +```java +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）迁移元素时会锁住当前桶，也是分段锁的思想； + +### 删除元素 + +删除元素跟添加元素一样，都是先找到元素所在的桶，然后采用分段锁的思想锁住整个桶，再进行操作。 + +```java +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()方法的重写。 + +```java +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）获取元素没有加锁； + +### 获取元素个数 + +元素个数的存储也是采用分段的思想，获取元素个数时需要把所有段加起来。 + +```java +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）获取元素个数没有加锁； + diff --git a/week_04/44/ConcurrentLinkedQueue.md b/week_04/44/ConcurrentLinkedQueue.md new file mode 100644 index 0000000000000000000000000000000000000000..9893efad4185fe5e5b4392c3d169296ed68dcecf --- /dev/null +++ b/week_04/44/ConcurrentLinkedQueue.md @@ -0,0 +1,211 @@ +## ConcurrentLinkedQueue + +## 参考 + +彤哥读源码 + +## 简介 + +​ 在并发编程中，有时候需要使用线程安全的队列。如果要实现一个线程安全的队列有两种方式：一种是使用阻塞算法，另一种是使用非阻塞算法。使用阻塞算法的队列可以用一个锁（入队和出队用同一把锁）或两个锁（入队和出队用不同的锁）等方式来实现。非阻塞的实现方式则可以使用循环CAS的方式来实现。 + +​ ConcurrentLinkedQueue是一个基于链接节点的无界线程安全队列，它采用先进先出的规则对节点进行排序，当我们添加一个元素的时候，它会添加到队列的尾部；当我们获取一个元素时，它会返回队列头部的元素。它采用了“wait-free”算法（即CAS算法）来实现，该算法在Michael&Scott算法上进行了一些修改。 + +​ ConcurrentLinkedQueue只实现了Queue接口，并没有实现BlockingQueue接口，所以它不是阻塞队列，也不能用于线程池中，但是它是线程安全的，可用于多线程环境中。 + +## 源码分析 + +### 属性 + +```java +// 链表头节点 +private transient volatile Node head; +// 链表尾节点 +private transient volatile Node tail; +``` + +### 主要内部类 + +```java +private static class Node { + volatile E item; + volatile Node next; +} +``` + +典型的单链表结构，非常纯粹。 + +### 主要构造方式 + +```java +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)方法也不用抛出异常了。 + +```java +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()。 + +```java +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/44/CopyOnWriteArrayList.md b/week_04/44/CopyOnWriteArrayList.md new file mode 100644 index 0000000000000000000000000000000000000000..480bfc4d4cab44b3c73cf1e844d5dee3d0efacad --- /dev/null +++ b/week_04/44/CopyOnWriteArrayList.md @@ -0,0 +1,267 @@ +# CopyOnWriteArrayList + +## 参考 + +彤哥读源码 + +## 简介 + +CopyOnWriteArrayList是ArrayList的线程安全版本，内部也是通过数组实现，每次对数组的修改都完全拷贝一份新的数组来修改，修改完了再替换掉老数组，这样保证了只阻塞写操作，不阻塞读操作，实现读写分离。 + +CopyOnWriteArrayList实现了List, RandomAccess, Cloneable, java.io.Serializable等接口。 + +CopyOnWriteArrayList实现了List，提供了基础的添加、删除、遍历等操作。 + +CopyOnWriteArrayList实现了RandomAccess，提供了随机访问的能力。 + +CopyOnWriteArrayList实现了Cloneable，可以被克隆。 + +CopyOnWriteArrayList实现了Serializable，可以被序列化。 + +## 源码分析 + +### 属性 + +```java +// 用于修改时加锁 +final transient ReentrantLock lock = new ReentrantLock(); +// 真正存储数据的地方，只能通过getArray()/setArray()访问 +private transient volatile Object[] array; +``` + +（1）lock + +用于修改时加锁，使用transient修饰表示不自动序列化。 + +（2）array + +真正存储元素的地方，使用transient修饰表示不自动序列化，使用volatile修饰表示一个线程对这个字段的修改另外一个线程立即可见。 + +### 构造方法 + +```java +final void setArray(Object[] a) { + array = a; +} +// 创建一个空数组 +public CopyOnWriteArrayList() { + setArray(new Object[0]); +} +// 把toCopyIn的元素拷贝给当前list的数组。 +public CopyOnWriteArrayList(E[] toCopyIn) { + setArray(Arrays.copyOf(toCopyIn, toCopyIn.length, Object[].class)); +} +/** 如果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 { + elements = c.toArray(); + // c.toArray might (incorrectly) not return Object + if (elements.getClass() != Object[].class) + elements = Arrays.copyOf(elements, elements.length, Object[].class); + } + setArray(elements); +} +``` + +### add(E e) 方法 + +添加一个元素到末尾。 + +```java +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(); + } +} +``` + +### add(int index, E element)方法 + +添加一个元素到指定位置 + +```java +public void add(int index, E element) { + final ReentrantLock lock = this.lock; + // 加锁 + lock.lock(); + try { + // 获取旧数组 + Object[] elements = getArray(); + int len = elements.length; + // 判断是否越界 + 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 { + // index不在最后一位，建一个n+1大小的空数组，拷贝旧数组前index个元素到新数组中 + newElements = new Object[len + 1]; + System.arraycopy(elements, 0, newElements, 0, index); + // 拷贝旧数组index及其之后的元素到新数组index+1开始的后面位置中 + System.arraycopy(elements, index, newElements, index + 1, + numMoved); + } + // 在index处加入要添加的值 + newElements[index] = element; + setArray(newElements); + } finally { + // 释放锁 + lock.unlock(); + } +} +``` + +### addIfAbsent(E e)方法 + +添加一个元素，如果这个元素不存在集合当中 + +```java +public boolean addIfAbsent(E e) { + // 获取旧数组 + Object[] snapshot = getArray(); + // 用indexOf判断元素是否存在于集合当中，不存在则添加 + 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(); + } +} +``` + +### get(int index)方法 + +获取指定索引的元素 + +```java +private E get(Object[] a, int index) { + return (E) a[index]; +} + +public E get(int index) { + return get(getArray(), index); +} + +final Object[] getArray() { + return array; +} +``` + +（1）获取元素数组； + +（2）返回数组指定索引位置的元素； + +（3）没有做越界检查，因为数组本身会做越界检查 + +### remove(int index)方法 + +删除指定位置的索引，并返回被删除元素。 + +```java +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) + // 如果被删除元素在最后一位，则拷贝除最后一位的元素到新数组 + setArray(Arrays.copyOf(elements, len - 1)); + else { + Object[] newElements = new Object[len - 1]; + // 拷贝索引前面的元素 + System.arraycopy(elements, 0, newElements, 0, index); + // 拷贝索引后面的元素 + System.arraycopy(elements, index + 1, newElements, index, + numMoved); + setArray(newElements); + } + // 返回旧值 + return oldValue; + } finally { + // 释放锁 + lock.unlock(); + } +} +``` + +### size()方法 + +```java +// 返回数组长度 +public int size() { + return getArray().length; +} +``` + +### 总结 + +（1）CopyOnWriteArrayList使用ReentrantLock重入锁加锁，保证线程安全； + +（2）CopyOnWriteArrayList的写操作都要先拷贝一份新数组，在新数组中做修改，修改完了再用新数组替换老数组，所以空间复杂度是O(n)，性能比较低下； + +（3）CopyOnWriteArrayList的读操作支持随机访问，时间复杂度为O(1)； + +（4）CopyOnWriteArrayList采用读写分离的思想，读操作不加锁，写操作加锁，且写操作占用较大内存空间，所以适用于读多写少的场合； + +（5）CopyOnWriteArrayList只保证最终一致性，不保证实时一致性； + +（6）为什么CopyOnWriteArrayList没有size属性？ + +因为每次修改都是拷贝一份正好可以存储目标个数元素的数组，所以不需要size属性了，数组的长度就是集合的大小，而不像ArrayList数组的长度实际是要大于集合的大小的。 \ No newline at end of file diff --git a/week_04/44/DelayQueue.md b/week_04/44/DelayQueue.md new file mode 100644 index 0000000000000000000000000000000000000000..d4481101d609bb03e2ab06ebb08f72f5f98feaa1 --- /dev/null +++ b/week_04/44/DelayQueue.md @@ -0,0 +1,88 @@ +# DelayQueue + +## 参考 + +彤哥读源码 + +《Java并发编程的艺术》 + +## 分析 + +​ DelayQueue是一个支持延时获取元素的无界阻塞队列。队列使用PriorityQueue来实现。队列中的元素必须实现Delayed接口，在创建元素时可以指定多久才能从队列中获取当前元素。只有在延迟期满时才能从队列中提取元素。 +DelayQueue非常有用，可以将DelayQueue运用在以下应用场景。 + +·缓存系统的设计：可以用DelayQueue保存缓存元素的有效期，使用一个线程循环查询DelayQueue，一旦能从DelayQueue中获取元素时，表示缓存有效期到了。 + +·定时任务调度：使用DelayQueue保存当天将会执行的任务和执行时间，一旦从DelayQueue中获取到任务就开始执行，比如TimerQueue就是使用DelayQueue实现的。 + +（1）如何实现Delayed接口 + DelayQueue队列的元素必须实现Delayed接口。我们可以参考ScheduledThreadPoolExecutor里ScheduledFutureTask类的实现，一共有三步。 + +第一步：在对象创建的时候，初始化基本数据。使用time记录当前对象延迟到什么时候可以使用，使用sequenceNumber来标识元素在队列中的先后顺序。代码如下。 + +```java +private static final AtomicLong sequencer = new AtomicLong(0); + ScheduledFutureTask(Runnable r, V result, long ns, long period) { + ScheduledFutureTask(Runnable r, V result, long ns, long period) { + super(r, result); + this.time = ns; + this.period = period; + this.sequenceNumber = sequencer.getAndIncrement(); +} +``` + +第二步：实现getDelay方法，该方法返回当前元素还需要延时多长时间，单位是纳秒，代码如下。 + +```java +public long getDelay(TimeUnit unit) { + return unit.convert(time - now(), TimeUnit.NANOSECONDS); +} +``` + +​ 通过构造函数可以看出延迟时间参数ns的单位是纳秒，自己设计的时候最好使用纳秒，因为实现getDelay()方法时可以指定任意单位，一旦以秒或分作为单位，而延时时间又精确不到纳秒就麻烦了。使用时请注意当time小于当前时间时，getDelay会返回负数。 + +第三步：实现compareTo方法来指定元素的顺序。例如，让延时时间最长的放在队列的末尾。实现代码如下。 + +```java +public int compareTo(Delayed other) { + if (other == this) // compare zero ONLY if same object + return 0; + if (other instanceof ScheduledFutureTask) { + ScheduledFutureTask<> x = (ScheduledFutureTask<>)other; + long diff = time - x.time; + if (diff < 0) + return -1; + else if (diff > 0) + return 1; + else if (sequenceNumber < x.sequenceNumber) + return -1; + else + return 1; + } + long d = (getDelay(TimeUnit.NANOSECONDS) - other.getDelay(TimeUnit.NANOSECONDS)); + return (d == 0) 0 : ((d < 0) -1 : 1); +} +``` + +（2）如何实现延时阻塞队列 +延时阻塞队列的实现很简单，当消费者从队列里获取元素时，如果元素没有达到延时时间，就阻塞当前线程。 + +```java +long delay = first.getDelay(TimeUnit.NANOSECONDS); +if (delay <= 0) + return q.poll(); +else if (leader != null) + available.await(); +else { + Thread thisThread = Thread.currentThread(); + leader = thisThread; + try { + available.awaitNanos(delay); + } finally { + if (leader == thisThread) + leader = null; + } +} +``` + +​ 代码中的变量leader是一个等待获取队列头部元素的线程。如果leader不等于空，表示已经有线程在等待获取队列的头元素。所以，使用await()方法让当前线程等待信号。如果leader等于空，则把当前线程设置成leader，并使用awaitNanos()方法让当前线程等待接收信号或等待delay时间。 \ No newline at end of file