ArrayBlockingQueue 迭代器

向右看齐 2022-11-21 03:38 26阅读 0赞

`ArrayBlockingQueue` 是一个多线程容器，但是为了方便读者只管的理解，本文所有理解都在单线程下做演示效果，但是多线程的结果是一样。

`ArrayBlockingQueue` 迭代器比较特殊：

1.  他不会像`HashMap`那样会出现`ModifyException`异常
2.  可以遍历出新插入的元素(前提是迭代前没有删除过全部元素的行为)
3.  会自动跳过其他线程删除的元素。(但是不会跳过第一个元素)
4.  如果迭代器和`ArrayBlockingQueue`相差两个循环操作以上，那么迭代器无效但是可以输出开始遍历第一个元素

我们首先来看第三个特性和第二个特性测试。读者可自动扩展多线程测试

ArrayBlockingQueue arrayBlockingQueue = new ArrayBlockingQueue(6, false);
    
            arrayBlockingQueue.put("1");
            arrayBlockingQueue.put("2");
            arrayBlockingQueue.put("3");
            arrayBlockingQueue.put("4");
    
            //故意构造迭代器在两个插入操作后
            Iterator iterator = arrayBlockingQueue.iterator();
            //取出两个元素
    
            arrayBlockingQueue.take();
            arrayBlockingQueue.take();
        	arrayBlockingQueue.put("5");
            arrayBlockingQueue.put("6");
    
    
    
            for (Iterator it = iterator; it.hasNext(); ) {
        
                Object next = it.next();
                //输出 next 1 不会输出 3
                System.out.println(" next " + next);
            }
            System.out.println("结束");

输出：

next 1
    next 3
    next 4
    next 5
    next 6
    结束

虽然`next1`被删除但是依然可以被读取。  
`next2`被删除所以被跳过。  
`next 5`和`next 6`是迭代前插入的元素（多线程遍历时同时插入也可以感知到，这里单线程只是方便演示）

我们看下什么时候不能读取到新的插入元素情况：  
(读者可自行扩展到多线程情况)

ArrayBlockingQueue arrayBlockingQueue = new ArrayBlockingQueue(7, false);
            arrayBlockingQueue.put("1");
            arrayBlockingQueue.put("2");
            arrayBlockingQueue.put("3");
            arrayBlockingQueue.put("4");
    
            //故意构造迭代器在两个插入操作后
            Iterator iterator = arrayBlockingQueue.iterator();
            //取出两个元素
    
            arrayBlockingQueue.take();
            arrayBlockingQueue.take();
            arrayBlockingQueue.take();
            arrayBlockingQueue.take();
            //前面删除了全部元素所以 5 和6 无法读取
            arrayBlockingQueue.put("5");
            arrayBlockingQueue.put("6");
    
            for (Iterator it = iterator; it.hasNext(); ) {
        
                Object next = it.next();
                //输出 next 1 不会输出 3
                System.out.println(" next " + next);
            }
            System.out.println("结束");

输出

next 1
    结束

第四个特征如果迭代器和`ArrayBlockingQueue`相差两个循环操作以上，那么迭代器无效但是可以输出开始遍历第一个元素

ArrayBlockingQueue arrayBlockingQueue = new ArrayBlockingQueue(7, false);
            arrayBlockingQueue.put("1");
            arrayBlockingQueue.put("2");
            arrayBlockingQueue.put("3");
            arrayBlockingQueue.put("4");
    
            Iterator iterator = arrayBlockingQueue.iterator();
            //注意ArrayBlockingQueue的capacity为7 遍历20次所以一定循环了两次列表
            for (int i = 0; i < 20; i++) {
        
                arrayBlockingQueue.put("test ");
                arrayBlockingQueue.take();
            }
            System.out.println("当前："+arrayBlockingQueue.size());
    
            for (Iterator it = iterator; it.hasNext(); ) {
        
                Object next = it.next();
                //输出 next 1 不会输出 3
                System.out.println(" next " + next);
            }
            System.out.println("结束");

输出

当前：4
     next 1
    结束

## 源码分析 ##

//ArrayBlockingQueue.java
      public Iterator<E> iterator() {
        
            return new Itr();
        }

//ArrayBlockingQueue.java 内部类
    Classs Itr{
        
    	Itr() {
        
            lastRet = NONE;
            final ReentrantLock lock = ArrayBlockingQueue.this.lock;
            lock.lock();
            try {
        
            	//如果当前没有元素 那么标识为空
                if (count == 0) {
        
                    cursor = NONE;
                    nextIndex = NONE;
                    prevTakeIndex = DETACHED;
                } else {
        
                	//ArrayBlockingQueue对象内部也有一个takeIndex，但是
                	//迭代器构造时存储当前ArrayBlockingQueue的takeIndex数值
                	//用于判断当前迭代器元素和ArrayBlockingQueue是否一致
                    final int takeIndex = ArrayBlockingQueue.this.takeIndex;
                    prevTakeIndex = takeIndex;
                    //nextIndex 主要在next函数使用。表示上次取出元素的下标
                    //nextItem 迭代器next函数返回的元素
                    nextItem = itemAt(nextIndex = takeIndex);
                    //incCursor主要用来校验 传入的下标是否合法
                    //如果合法应该大于0的数值，cursor如果合法那么为下一次遍历的数组下标
                   	//可以简单理解为cursor对于nextIndex而已下一次遍历的下标。
                   	//此属性在每次在next函数中取出元素后加一 
                    cursor = incCursor(takeIndex);
                    //itrs对象存储多个迭代器对象实例
                    //用来同步集合ArrayBlockingQueue状态 被删除元素后，进行的状态同步
                    if (itrs == null) {
        
                    	//构造一个实例并把当前迭代器放入
                        itrs = new Itrs(this);
                    } else {
        
                    	//注册当前迭代器到itrs中
                        itrs.register(this); // in this order
                        //清楚无效的迭代器
                        itrs.doSomeSweeping(false);
                    }
                    //注意cycle表示队列所有数组元素都被取出过一次次数
                    //换句话如果下次取出的元素的下标为0，cycles就+1，请深刻理解。
                    prevCycles = itrs.cycles;
                  
                }
            } finally {
        
                lock.unlock();
            }
        }
        
        //校验index是否合法，如果合法返回原始值
        //如果大于0那么合法的
        private int incCursor(int index) {
        
     			//下一个遍历下标如果到底数组最大值，那么重置到0	
                if (++index == items.length)
                    index = 0;
                //如果遍历的下标等于下一次ArrayBlockingQueue放入元素的位置，证明遍历完成.由于迭代器时内部类所以可以读取ArrayBlockingQueue的putIndex
                if (index == putIndex)
                    index = NONE;
                return index;
            }
         //返回元素
         final E itemAt(int i) {
        
            return (E) items[i];
         }
     }

其中上面有一个`itrs`的属性是`ArrayBlockingQueue`属性，用于同步迭代器和`ArrayBlockingQueue`，方便跳过已读取和标志迭代器无效等。其内部会封装一个Node对象包装迭代器，而存储多个迭代器采用链表结构

//ArrayBlockingQueue.java
    
    class Iters{
        
      //方便构造链表存储迭代器
      private class Node extends WeakReference<Itr> {
        
                Node next;
    
                Node(Itr iterator, Node next) {
        
                    super(iterator);
                    this.next = next;
                }
            }
    	//同步ArrayBlockingQueue状态		
    	 void elementDequeued() {
        
    	 		//如果某次删除元素后，整个ArrayBlockingQueue数组为空，
    	 		//所以当迭代前删除过一次全部元素，会导致无法读取新插入的元素
                if (count == 0)
                    queueIsEmpty();
                    //takeIndex表示ArrayBlockingQueue下一次取出的元素下标
                    //takeIndex == 0 表示 ArrayBlockingQueue 经过多次操作后
                    //又回到取出又回到数组0位置上，那么让cycles自增1 ,然后通知所有
                    //迭代器检查是否需改更正自己的状态 为无效迭代器，防止遍历
    
                else if (takeIndex == 0)
                   //调用Itrs的takeIndexWrapped函数让cycles自增1
                   //并且通知其管理的迭代器检查自己是否需要更新状态 
                    takeIndexWrapped();
            }
    	//通知所有迭代器全部无效，不需要在迭代了
    	void queueIsEmpty() {
        
                for (Node p = head; p != null; p = p.next) {
        
                    Itr it = p.get();
                    if (it != null) {
        
                  		//清空Node存储的迭代器对象
                        p.clear();
                        //标志迭代器一些变量无效
                        it.shutdown();
                    }
                }
                
                //head是一个Node对象实例，Node是一个链表节点
                //Node指向迭代器
                head = null;
                //itrs是`ArrayBlockingQueue`内部属性
                itrs = null;
            }
    }

一些标志函数

//ArrayBlockingQueue.java 内部类
    class Itr{
        
    		//简单理解true 为当前迭代器无效
            boolean isDetached() {
        
            	//prevTakeIndex 正常指向takeIndex
            	//构造迭代时赋值
                return prevTakeIndex < 0;
            }
            //这个函数的作用如下几点
            // 1 迭代器如果是不能再用了 那么进行清理资源
            // 2 因为ArrayBlockingQueue取出元素导致迭代器需要更新一些信息，以方便跳过一些被删除的元素
            // 3 标记一些特殊情况导致迭代器不能使用，让其停止遍历
            private void incorporateDequeues(){
        
            //...后面在详解
            }
    }

迭代器常规函数，`hasNext()`判断是否还有元素，`next()`返回元素

class Itr{
        
    	public boolean hasNext() {
        
    					//直接判断nextItem
    					//nextItem指向的是下一次next函数返回元素
    		            if (nextItem != null)
    		                return true;
    		            //如果运行到证明迭代器已经返回false，没有元素在能变能遍历
    		            //所以清除资源标记无效等
    		            noNext();
    		            return false;
    		        }
    
    	private void noNext() {
        
                final ReentrantLock lock = ArrayBlockingQueue.this.lock;
                lock.lock();
                try {
        
                //当前迭代器是否还有效，true无效
                //!isDetached() 表示当前迭代器还有效
                    if (!isDetached()) {
        
                    	//标记无效清理资源等
                        incorporateDequeues(); 
                        if (lastRet >= 0) {
        
                            lastItem = itemAt(lastRet);
                            detach();
                        }
                    }
                } finally {
        
                    lock.unlock();
                }
            }
    
    	public E next() {
        
    			//要返回的元素
                final E x = nextItem;
                if (x == null)
                    throw new NoSuchElementException();
                final ReentrantLock lock = ArrayBlockingQueue.this.lock;
                lock.lock();
                try {
        
                	//当前迭代器是否还有效，true无效
                    if (!isDetached())
                    	//同步状态跳过多线程读取过的元素
                    	//或者标记迭代器无效
                        incorporateDequeues();
                    
                    
                    lastRet = nextIndex;
                    final int cursor = this.cursor;
                    //直接取下一次迭代的nextItem
                    if (cursor >= 0) {
        
                        nextItem = itemAt(nextIndex = cursor);
                        //cursor计数器+1，读取下一个元素的坐标
                        // 可以简答理解为  数组长度%++cursor
                        this.cursor = incCursor(cursor);
                    } else {
        
                    	//cursor小于0证明迭代器被标记为为无效了
                    	//incCursor等函数都可以标记cursor小于0
                    	//nextItem等于null 那么hasNext会返回false
                        nextIndex = NONE;
                        nextItem = null;
                    }
                } finally {
        
                    lock.unlock();
                }
                return x;
            }
    }

我们看下`ArrayBlockingQueue`删除元素时候进行的同步的代码：

//ArrayBlockingQueue.java
    class ArrayBlockingQueue{
        
      private E dequeue() {
        
      		//常规删除元素代码
            final Object[] items = this.items;
            E x = (E) items[takeIndex];
            items[takeIndex] = null;
            if (++takeIndex == items.length)
                takeIndex = 0;
            //count表示存储的元素数量
            count--;
            
            //同步itrs注册的迭代器
            if (itrs != null)
            	//上文有次函数讲解
                itrs.elementDequeued();
            
            notFull.signal();
            return x;
        }
    }

class Iters{
        
    //同步ArrayBlockingQueue状态		
    	 void elementDequeued() {
        
    	 		//队列为空标志所有迭代器无效
                if (count == 0)
                    queueIsEmpty();
                    //takeIndex == 0循环队列经过一次循环
                else if (takeIndex == 0)
                   //调用Itrs的takeIndexWrapped函数让cycles自增1
                   //并且通知其管理的迭代器检查自己是否需要更新状态 
                    takeIndexWrapped();
            }
    }

//ArrayBlockingQueue.java
    class Itrs{
        
     void takeIndexWrapped() {
        
     			//这里指代的是经过多次取出和删除，数组取出索引take
                cycles++;
                //通知
                for (Node o = null, p = head; p != null;) {
        
                    final Itr it = p.get();
                    final Node next = p.next;
                    //takeIndexWrapped 返回true标识迭代器已经不可用了
                    //从链表清楚
                    if (it == null || it.takeIndexWrapped()) {
        
                        //清空Node存储的迭代器的引用
                        p.clear();
                        //移动到下一级链表操作
                        p.next = null;
                        if (o == null)
                            head = next;
                        else
                            o.next = next;
                    } else {
        
                        o = p;
                    }
                    p = next;
                }
                if (head == null)   // no more iterators to track
                    itrs = null;
            }
    }

class Itr{
        
    
    	boolean takeIndexWrapped() {
        
    			//当前已经无效了直接返回
                if (isDetached())
                    return true;
                //迭代器记录的循环次数和当前的循环次数大于等于2的话
                //直接废弃迭代器为无效
                //这里为什么是大于等于2才废弃，
                //第一个笔者认为jdk编写者相差过大
      			//另外一个如果相差小于等于1，那么其实可以跳过被删除的元素迭代器继续遍历
                if (itrs.cycles - prevCycles > 1) {
        
                   //标记当前迭代器无效
                    shutdown();
                    return true;
                }
                return false;
            }
            
         
            //设置所有标志无效
            //但是nextItem不会设置为null,因这个函数会并发调用所以有可能其他线程正在使用迭代器调用 next函数 ，如果为null可能会引起空指针异常
            //但是这个也会带来一个弊端，shutdown后迭代器会残留一次脏数据。      
     		void shutdown() {
        
                cursor = NONE;
                if (nextIndex >= 0)
                    nextIndex = REMOVED;
                if (lastRet >= 0) {
        
                    lastRet = REMOVED;
                    lastItem = null;
                }
                prevTakeIndex = DETACHED;
                // Don't set nextItem to null because we must continue to be
                // able to return it on next().
                //
                // Caller will unlink from itrs when convenient.
            }
    }

我们在回头看下如何状态同步呢。

`incorporateDequeues`用于`next`函数`ArrayBlockingQueue`同步状态。

class Itr{
        
    	private void incorporateDequeues() {
        
               //队列 被循环使用过几次
               //每次队列中下标为0的地方取元素就加1
                final int cycles = itrs.cycles;
                
                final int takeIndex = ArrayBlockingQueue.this.takeIndex;
                //迭代器存储 的循环次数
                final int prevCycles = this.prevCycles;
                //构造迭代器的时候传入的取出元素的下标	
                final int prevTakeIndex = this.prevTakeIndex;
    			
    			//证明迭代器和ArrayBlockingQueue状态不一致
                if (cycles != prevCycles || takeIndex != prevTakeIndex) {
        
                    final int len = items.length;
                    //得到迭代器和ArrayBlockingQueue到底存在多少次迭代差别区间
                    //比如当前ArrayBlockingQueue下次要取出 下标为10，而
                    //迭代器还在取下标为3的
                    long dequeues = (cycles - prevCycles) * len
                        + (takeIndex - prevTakeIndex);
    
                    // lastRet指代上一次迭代返回元素下标
                    if (invalidated(lastRet, prevTakeIndex, dequeues, len))
                        lastRet = REMOVED;
                     // nextIndex大多数情况和lastRet一样
                    if (invalidated(nextIndex, prevTakeIndex, dequeues, len))
                        nextIndex = REMOVED;
                    //cursor表示下一次next取出元素的下标
                    if (invalidated(cursor, prevTakeIndex, dequeues, len))
                    	//读取下一个元素的下标更新。也就是会跳过被删除的元素
                        cursor = takeIndex;
                        
    				//如果无法恢复那么直接干脆不同步了
                    if (cursor < 0 && nextIndex < 0 && lastRet < 0)
                    	//从itrs移除当前迭代器
                        detach();
                    else {
        
                    	//虽然ArrayBlockingQueue和迭代器的状态不一致
                    	//但是在可允许的范围内所以更新迭代器的状态
                        this.prevCycles = cycles;
                        this.prevTakeIndex = takeIndex;
                    }
                }
            }
    
    	// index数据和ArrayBlockingQueue数据不一致导致陈旧数据 就返回true
    	   private boolean invalidated(int index, int prevTakeIndex,
                                        long dequeues, int length) {
        
                
    			//prevTakeIndex 表示构造迭代器的时候存储ArrayBlockingQueue下次take取出元素的下标
    			
    
                //dequeues表示ArrayBlockingQueue和迭代器存储的状态相差的状态步数
                //比如ArrayBlockingQueue在迭代器构造后take 一次就+1
                
    			
    			//length 数组长度
    			
    			//数据无效直接返回false
                if (index < 0)
                    return false;
                 
                //存储初始状态下标和现在数据的差值，表示遍历了多少次
                //差值为负数:
                //			(1) index 已经从数组尾部遍历到数组头部，所以需要加上长度修正
                int distance = index - prevTakeIndex;
    			
    			//举个例子 prevTakeIndex =3 length = 4 index 0
    			// index - prevTakeIndex = -3 
    			// distance += length = 1
                if (distance < 0)
                    distance += length;
              
              //ArrayBlockingQueue和迭代器存储的状态相差步数 大于迭代器遍历次数
              //那么证明必然存在数据变化，会导致不一样数据
              //举个例子： 数组元素为： | a | b | c | d | e |
              // 迭代器从 d 遍历到 e ，那么distance为 1
              // 而ArrayBlockingQueue 被take 3次 数组元素变为    | null | b | c  | null | null |
              //如果发生上面的情况迭代不能再继续向下迭代，需要更正
              //如果迭代器步数大于ArrayBlockingQueue那么其实不影响继续向下迭代
                return dequeues > distance;
            }
    }        
    }