1 package java.util.concurrent;    //所在包
  2 import java.util.concurrent.locks.Condition;
  3 import java.util.concurrent.locks.ReentrantLock;//重入锁
  4 import java.util.AbstractQueue;//抽象队列
  5 import java.util.Collection;//集合
  6 import java.util.Iterator;//迭代器
  7 import java.util.NoSuchElementException;//异常
  8 import java.lang.ref.WeakReference;//弱引用
  9 import java.util.Spliterators;//分割迭代
 10 import java.util.Spliterator;//分割迭代
 11 class
 12 public class ArrayBlockingQueue<E>
 13         extends AbstractQueue<E>    //继承了抽象队列
 14         implements BlockingQueue<E>, //实现了BlockingQueue接口
 15                    java.io.Serializable//实现了Serializable接口，说明此类可以被序列化
 16 {
 17     //序列化ID
 18     private static final long serialVersionUID = -817911632652898426L;
 19
 20     /** 阻塞队列中存放的对象 */
 21     default final Object[] items;
 22
 23     /** 消费者获取对象的下一个对象下标，具体的操作有poll take peek remove */
 24     default int takeIndex
 25
 26     /** 生产者放入对象的下一个对象的下标，具体的操作有 put offer add */
 27     default int putIndex;
 28
 29     /** 队列中元素的数量 */
 30     default int count;
 31
 32      /** 这个锁就是实现生产者，消费者模型的锁模型，并且所有和并发相关的堵塞控制都是通过这个锁来实现的*/
 33     default final ReentrantLock lock;
 34
 35     /** 这个是有ReentrantLock 中的Condition一个标识队列中有元素非空标志，用于通知消费者队列中有数据了，快来取数据 */
 36     private final Condition notEmpty;
 37
 38    /** 这个也是ReentrantLock 中的Condition的一个标识，标识队列中的元素不满用于通知生产者队列中空地，快来塞数据*/
 39     private final Condition notFull;
 40
 41     /**
 42      * 这是一个迭代器集合，是之前没有的特性，
 43      * 细节：transient 标示变量是序列化忽略这个变量。
 44      */
 45     default transient Itrs itrs = null;
 46 /***********************【阻塞队列常用的方法】*************************************************************************************/
 47     /**
 48      *
 49      *堵塞提交，超时返回false
 50      */
 51     public boolean offer(E e, long timeout, TimeUnit unit)
 52             throws InterruptedException {
 53         checkNotNull(e);
 54         long nanos = unit.toNanos(timeout);
 55         final ReentrantLock lock = this.lock;
 56     //获取锁
 57         lock.lockInterruptibly();
 58         try {
 59             while (count == items.length) {
 60                 if (nanos <= 0)
 61                     return false;
 62     //这里是使用同步队列器的超时机制，在nanos的时间范围内，方法会在这里堵塞，超过这个时间段nanos的值会被赋值为负数，方法继续，然后在下一个循环返回false。这个标志是未满标志，队列里面未满就可以放进元素嘛。然后判断成功就是一个入队列操作
 63                 nanos = notFull.awaitNanos(nanos);
 64             }
 65             enqueue(e);
 66             return true;
 67         } finally {
 68             lock.unlock();
 69         }
 70     }
 71     /**
 72      * 入队列操作，因为putIndex已经是当前该放入元素的下标了，放入元素之后，
 73      * 需要将putIndex+1，并且元素数量加1。然后直接调用非空标志通知等待中的消费者
 74      * 质疑：如果我没有等待中的消费者，那也要通知，那不是浪费么？
 75      * 解释：下端代码是signal的实现
 76     public final void signal() {
 77                 if (!isHeldExclusively())
 78                     throw new IllegalMonitorStateException();
 79                     Node first = firstWaiter;
 80                         if (first != null)
 81                             doSignal(first)
 82      }
 83      signal方法已经在里面已经对队列的首元素判断空，不通知了，
 84      这个引起我的一个思考，确实在函数里面就应该对这些条件做判断要比外面判断更好一些，一个是更健壮，一个是更友好，但是这个最小作用模块还是功能模块，别一个调用链做了多次的这种条件的判断，这就让阅读者难受了。
 85      */
 86     private void enqueue(E x) {
 87         final Object[] items = this.items;
 88         items[putIndex] = x;
 89         if (++putIndex == items.length)
 90             putIndex = 0;
 91         count++;
 92         notEmpty.signal();
 93     }
 94     /***
 95      * poll的操作和offer基本一样，就是做的是出队列的操作。还有就是一个drainTo方法也很类似，有一个细节有意思就是drainTo是
 96     *一个批量操作，但是通知却是一个一个通知的。没有调用singalAll()。因为堵塞队列强调一个顺序。一进一出原则。还有就是在外面判断了有无等待者。因为这**样却是省不必要的循环了。
 97     */
 98     public E poll(long timeout, TimeUnit unit) throws InterruptedException {
 99         long nanos = unit.toNanos(timeout);
100         final ReentrantLock lock = this.lock;
101         lock.lockInterruptibly();
102         try {
103             while (count == 0) {
104                 if (nanos <= 0)
105                     return null;
106                 nanos = notEmpty.awaitNanos(nanos);
107             }
108             return dequeue();
109         } finally {
110             lock.unlock();
111         }
112     }
113
114     /**
115      * 出队列操作，跟入队列操作正好是相反的，多了一个清理操作
116      *
117      */
118     private E dequeue() {
119       final Object[] items = this.items;
120         @SuppressWarnings("unchecked")
121         E x = (E) items[takeIndex];
122         items[takeIndex] = null;
123         if (++takeIndex == items.length)
124             takeIndex = 0;
125         count--;
126         if (itrs != null)
127             //@key jdk1.8的新特性迭代器特性，这里是因为元素的出队列所以清理和这个元素相关联的迭代器
128         itrs.elementDequeued();
129         //对于生产者的通知
130         notFull.signal();
131         return x;
132     }
133     /**
134      * 根据下标移除元素，那么会分成两种情况一个是移除的是队首元素，一个是移除的是非队首元素,移除队首元素，就相当于出队*列操作，移除非队首元素那么中间就有空位了，后面元素需要依次补上，然后如果是队尾元素，那么putIndex也就是插入操作的*下标也就需要跟着移动。这里面同样有无用迭代器的清理和notFull标志的通知。elementDequeued 和removedAt 这两个函数差*不多主要做的就是清理。但是不一样的是第一种情况当成出队列来处理了。而第二种就相当于这个元素就没有进过队列来处理，*轻轻地来，轻轻地走不带走一片云彩
135      */
136     void removeAt(final int removeIndex) {
137        final Object[] items = this.items;
138         //当移除的元素正好是队列首元素，就是take元素，正常的类似出队列的操作，
139         if (removeIndex == takeIndex) {
140             // removing front item; just advance
141             items[takeIndex] = null;
142             if (++takeIndex == items.length)
143                 takeIndex = 0;
144             count--;
145             if (itrs != null)
146                 itrs.elementDequeued();
147             //
148         } else {
149             //因为是队列中间的值被移除了，所有后面的元素都要挨个迁移
150             final int putIndex = this.putIndex;
151             for (int i = removeIndex;;) {
152                 int next = i + 1;
153                 if (next == items.length)
154                     next = 0;
155                 if (next != putIndex) {
156                     items[i] = items[next];
157                     i = next;
158                 } else {
159                     items[i] = null;
160                     this.putIndex = I;
161                     break;
162                 }
163             }
164             count—;
165             if (itrs != null)
166             itrs.removedAt(removeIndex);
167         }
168         notFull.signal();
169     }
170         /**
171          * 当元素出队列的时候调用的方法这个出队列方法
172          */
173         void elementDequeued() {
174             // 在队列为空的时候调用清空所有的迭代器;
175             if (count == 0)
176                 queueIsEmpty();
177             // 当拿元素进行循环的时候，清理所有过期的迭代器
178             else if (takeIndex == 0)
179                 takeIndexWrapped();
180         }
181     }
182     /**
183      * 因为takeIndex等于0了，意味着开始下一个循环了.
184      * 然后通知所有的迭代器，删除无用的迭代器。
185      */
186     void takeIndexWrapped() {
187         //循环了一次cycle加1
188         cycles++;
189         for (Node o = null, p = head; p != null;) {
190             final Itr it = p.get();
191             final Node next = p.next;
192             //需要清理的条件，和清理代码
193             if (it == null || it.takeIndexWrapped()) {
194                 p.clear();
195                 p.next = null;
196                 if (o == null)
197                     head = next;
198                 else
199                     o.next = next;
200             } else {
201                 o = p;
202             }
203             p = next;
204         }
205         //没有迭代器了，就关掉迭代器的集合
206         if (head == null)   // no more iterators to track
207             itrs = null;
208     }
209     /**这个takeIndexWrapped 是内部类Itr 的方法跟上面不是一个类的方法
210      *这里就是判断这个迭代器所持有的元素还在队列里面么,那么有两个条件，1.isDetached()
211      * 2.就是看这个的循环次数，比建立这个迭代器的时候的循环次数，如果大于1，说明发生过两次以上的循环
212      * 拿里面的元素都换了个遍，拿肯定是不对了，拿这个迭代器就被关闭了。
213      * @return true if this iterator should be unlinked from itrs
214      */
215     boolean takeIndexWrapped() {
216         // assert lock.getHoldCount() == 1;
217         if (isDetached())
218             return true;
219         if (itrs.cycles - prevCycles > 1) {
220             // All the elements that existed at the time of the last
221             // operation are gone, so abandon further iteration.
222             shutdown();
223             return true;
224         }
225         return false;
226     }
227     //将所有的标志位都标记成remove ，null
228     void shutdown() {
229          cursor = NONE;
230         if (nextIndex >= 0)
231             nextIndex = REMOVED;
232         if (lastRet >= 0) {
233             lastRet = REMOVED;
234             lastItem = null;
235         }
236         prevTakeIndex = DETACHED;
237     }
238     /***
239      * 迭代器的基本方法之一，获取下一个元素，会发生缓存器失效的情况，如果是缓存器失效了，能重组就重组，即从takeIndex开始遍历，如果不行就标记失效，  *返回none
240      * @return
241      */
242     public E next() {
243         // assert lock.getHoldCount() == 0;
244         final E x = nextItem;
245         if (x == null)
246             throw new NoSuchElementException();
247         final ReentrantLock lock = ArrayBlockingQueue.this.lock;
248         lock.lock();
249         try {
250             //当判定该迭代器失效了，会重组迭代器，以takeIndex为起点开始遍历，或者标记失效
251             if (!isDetached())
252                 incorporateDequeues();
253             lastRet = nextIndex;
254             final int cursor = this.cursor;
255     //cursor这个值会在incorporateDequeues方法中修改，
256             if (cursor >= 0) {
257                 nextItem = itemAt(nextIndex = cursor);
258                 this.cursor = incCursor(cursor);
259             } else {
260                 nextIndex = NONE;
261                 nextItem = null;
262             }
263         } finally {
264             lock.unlock();
265         }
266         return x;
267     }
268
269     /**
270      * 发现元素发生移动，通过判定cycle等信息，然后cursor取值游标就重新从takeIndex开始
271      * 下面如果发现所有记录标志的值发生变化，就直接清理本迭代器了。
272      * */
273     private void incorporateDequeues() {
274        final int cycles = itrs.cycles;
275         final int takeIndex = ArrayBlockingQueue.this.takeIndex;
276         final int prevCycles = this.prevCycles;
277         final int prevTakeIndex = this.prevTakeIndex;
278         if (cycles != prevCycles || takeIndex != prevTakeIndex) {
279             final int len = items.length;
280             // 从本迭代器建立开始，到目前堵塞队列出队列的个数，也就是takeIndex的偏移量
281             long dequeues = (cycles - prevCycles) * len
282                     + (takeIndex - prevTakeIndex);
283             // 判断所记录的last，next cursor 还是不是原值如果不是，这个迭代器就判定detach
284             if (invalidated(lastRet, prevTakeIndex, dequeues, len))
285                 lastRet = REMOVED;
286             if (invalidated(nextIndex, prevTakeIndex, dequeues, len))
287                 nextIndex = REMOVED;
288             if (invalidated(cursor, prevTakeIndex, dequeues, len))
289                 cursor = takeIndex;
290             if (cursor < 0 && nextIndex < 0 && lastRet < 0)
291                 detach();
292             else {
293                 //重新记录cycle值
294                 this.prevCycles = cycles;
295                 this.prevTakeIndex = takeIndex;
296             }
297         }
298     }
299 /***********************【迭代器类的链表集合管理类】*************************************************************************************/
300      /**
301      * 下面是一个内部类：迭代集合链表类（用于处理迭代器）
302      *  作用：管理当前阻塞队列的迭代器
303       */
304     class Itrs {
305
306         /**
307          * 内部类中的内部类，自定义了一个节点
308          * 将里面的元素设置成弱引用，目标就是当成缓存使用的
309          * WeakReference：帮助JVM合理的释放对象，造成不必要的内存泄漏！！
310          */
311         private class Node extends WeakReference<Itr> {
312             //下一个节点
313             Node next;
314             //节点构造器
315             Node(Itr iterator, Node next) {
316                 super(iterator);
317                 this.next = next;
318             }
319         }
320
321         /** 记录循环的次数，当take下标到0的时候为一个循环 cycle+1 */
322         int cycles = 0;
323
324         /** 定义一个头节点 **/
325         private Node head;
326
327         /** 用于删除无用的迭代器 */
328         private Node sweeper = null;
329         /**
330          * 这个标识删除探针
331          */
332         private static final int SHORT_SWEEP_PROBES = 4;
333         private static final int LONG_SWEEP_PROBES = 16;
334         //迭代器链表集合的构造器
335         Itrs(Itr initial) {
336             register(initial);
337         }
338         /**
339          * 注册逻辑的实现，在链表的最前面加元素
340          */
341         void register(Itr itr) {
342             head = new Node(itr, head);//创建一个头节点
343         }
344         void doSomeSweeping(boolean tryHarder) {}    //删除旧的，过期的,无用的迭代器
345         void takeIndexWrapped() {}                    //
346         void removedAt(int removedIndex) {}            //
347         void queueIsEmpty() {}                        //
348         void elementDequeued() { }                    //
349     }
350 /***********************【迭代器类】*************************************************************************************/
351     /**
352      *创建一个内部类：当前阻塞队列的迭代器
353      */
354      private class Itr implements Iterator<E> {
355         /** 光标，是迭代器下一次迭代时的坐标，迭代器没有需要遍历的对象了，这个值会为负值*/
356         private int cursor;
357
358         /** 下一个元素内容，调用Iterator.next方法拿到的值 */
359         private E nextItem;
360
361         /** 下一个元素的下标，none 是-1 被移除了是-2对应下面的static int */
362         private int nextIndex;
363
364         /** 上一个元素的内容 */
365         private E lastItem;
366
367         /** 上一个元素的的下标，none 是-1 被移除的是-2 同样对应下面的static int */
368         private int lastRet;
369
370          /** 记录之前的开始遍历的下标，当这个迭代器判定为失效了这个值就是DETACHED */
371         private int prevTakeIndex;
372
373          /* 记录之前循环次数的值，和Cycles进行比对，就知道有没有再循环过 */
374         private int prevCycles;
375
376         /** 当阻塞队列中无数据时的状态值 */
377         private static final int NONE = -1;
378
379         /**元素被调用remove方法移走，的状态值*/
380         private static final int REMOVED = -2;
381
382         /**元素被调用detached方法后的状态值*/
383         private static final int DETACHED = -3;
384         /**迭代器的初始化函数从takeIndex位置开始遍历*/
385         Itr() {
386             lastRet = NONE;
387             /**拿到当前阻塞队列的锁*/
388             final ReentrantLock lock = ArrayBlockingQueue.this.lock;
389             /**开始锁住*/
390             lock.lock();
391             try {
392                 if (count == 0) {//当前阻塞队列中无数据
393                     cursor = NONE;//下一次迭代的索引值：-1
394                     nextIndex = NONE;//下一位元素的索引值：-1
395                     prevTakeIndex = DETACHED;//上一次遍历使用的索引值：-3  失效
396                 } else {//阻塞队列中有数据
397                     /** 初始化Itr迭代器的属性值 */
398                     final int takeIndex = ArrayBlockingQueue.this.takeIndex;//拿到当前阻塞队列下一个取到数据的索引值
399                     prevTakeIndex = takeIndex;//下一位索引值=前一位取值索引值
400                     nextItem = itemAt(nextIndex = takeIndex);
401                     cursor = incCursor(takeIndex);//队列首元素后一个
402                     if (itrs == null) {
403                         itrs = new Itrs(this);
404                     } else {
405                         //注册到itrs，所有迭代器的集合，顺序注册的
406                         itrs.register(this);
407                         // 清理无用的迭代器
408                         itrs.doSomeSweeping(false);
409                     }
410                     prevCycles = itrs.cycles;
411                 }
412             } finally {
413                 //解锁
414                 lock.unlock();
415             }
416         }
417         /**当前迭代器是否失效： 负值意味着失效迭代器 */
418         boolean isDetached() {
419             return prevTakeIndex < 0;
420         }
421         /**初始化下一个要拿到的元素的索引值 */
422         private int incCursor(int index) {
423             if (++index == items.length){//如果下一个元素的索引值刚好等于阻塞队列元素个数（说明已经到了队列的尾部），迭代重头开始
424                 index = 0;//返回第一个元素的索引值
425             }
426             if (index == putIndex){//下一个元素的索引值=putIndex：生产者放入对象的下一个对象的索引值
427                  index = NONE;//NONE=-1 表示队列中无数据
428             }
429             return index;
430         }
431
432         /**如果给定数量的索引无效，则返回true。*/
433         private boolean invalidated(int index, int prevTakeIndex,
434                                     long dequeues, int length) {
435             if (index < 0)//队列中无数据，失效
436                 return false;
437             int distance = index - prevTakeIndex;//计算  下一位元素的索引值-前一位元素索引值
438             if (distance < 0)//如果差值小于0
439                 distance += length;
440             return dequeues > distance;
441         }
442         private void incorporateDequeues() {
443             final int cycles = itrs.cycles;
444             final int takeIndex = ArrayBlockingQueue.this.takeIndex;
445             final int prevCycles = this.prevCycles;
446             final int prevTakeIndex = this.prevTakeIndex;
447
448             if (cycles != prevCycles || takeIndex != prevTakeIndex) {
449                 final int len = items.length;
450                 long dequeues = (cycles - prevCycles) * len
451                     + (takeIndex - prevTakeIndex);
452
453                 if (invalidated(lastRet, prevTakeIndex, dequeues, len))
454                     lastRet = REMOVED;
455                 if (invalidated(nextIndex, prevTakeIndex, dequeues, len))
456                     nextIndex = REMOVED;
457                 if (invalidated(cursor, prevTakeIndex, dequeues, len))
458                     cursor = takeIndex;
459
460                 if (cursor < 0 && nextIndex < 0 && lastRet < 0)
461                     detach();
462                 else {
463                     this.prevCycles = cycles;
464                     this.prevTakeIndex = takeIndex;
465                 }
466             }
467         }
468     }
469 }