java并发之阻塞队列LinkedBlockingQueue与ArrayBlockingQueue

            public
            interface BlockingQueue<E> extends Queue<E> {

    //将指定的元素插入到此队列的尾部（如果立即可行且不会超过该队列的容量）
    //在成功时返回 true，如果此队列已满，则抛IllegalStateException。 boolean add(E e); 

    //将指定的元素插入到此队列的尾部（如果立即可行且不会超过该队列的容量） 
    // 将指定的元素插入此队列的尾部，如果该队列已满， 
    //则在到达指定的等待时间之前等待可用的空间,该方法可中断 boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException; 

    //将指定的元素插入此队列的尾部，如果该队列已满，则一直等到（阻塞）。 void put(E e) throws InterruptedException; 

    //获取并移除此队列的头部，如果没有元素则等待（阻塞）， 
    //直到有元素将唤醒等待线程执行该操作 
    E take() throws InterruptedException; 

    //获取并移除此队列的头部，在指定的等待时间前一直等到获取元素， //超过时间方法将结束
    E poll(long timeout, TimeUnit unit) throws InterruptedException; 

    //从此队列中移除指定元素的单个实例（如果存在）。 boolean remove(Object o); 
}

    //除了上述方法还有继承自Queue接口的方法 
    //获取但不移除此队列的头元素,没有则跑异常NoSuchElementException     E element(); 

    //获取但不移除此队列的头；如果此队列为空，则返回 null。     E peek(); 

    //获取并移除此队列的头，如果此队列为空，则返回 null。 
    E poll();

            public
            class ArrayBlockingQueue<E> extends AbstractQueue<E>
        implements BlockingQueue<E>, java.io.Serializable {

    /** 存储数据的数组 */final Object[] items;

    /**获取数据的索引，主要用于take，poll，peek，remove方法 */int takeIndex;

    /**添加数据的索引，主要用于 put, offer, or add 方法*/int putIndex;

    /** 队列元素的个数 */int count;


    /** 控制并非访问的锁 */final ReentrantLock lock;

    /**notEmpty条件对象，用于通知take方法队列已有元素，可执行获取操作 */privatefinal Condition notEmpty;

    /**notFull条件对象，用于通知put方法队列未满，可执行添加操作 */privatefinal Condition notFull;

    /**
       迭代器
     */transient Itrs itrs = null;
//构造方法源码public ArrayBlockingQueue(int capacity) {
     this(capacity, false);
 }

public ArrayBlockingQueue(int capacity, boolean fair) {
     if (capacity <= 0)
         thrownew IllegalArgumentException();
     this.items = new Object[capacity];
     lock = new ReentrantLock(fair);
     notEmpty = lock.newCondition();
     notFull =  lock.newCondition();
 }

//add方法实现，间接调用了offer(e)publicboolean add(E e) {
        if (offer(e))
            returntrue;
        elsethrownew IllegalStateException("Queue full");
    }

//offer方法publicboolean offer(E e) {
     checkNotNull(e);//检查元素是否为nullfinal ReentrantLock lock = this.lock;
     lock.lock();//加锁try {
         if (count == items.length)//判断队列是否满returnfalse;
         else {
             enqueue(e);//添加元素到队列returntrue;
         }
     } finally {
         lock.unlock();
     }
 }

//入队操作privatevoid enqueue(E x) {
    //获取当前数组final Object[] items = this.items;
    //通过putIndex索引对数组进行赋值
    items[putIndex] = x;
    //索引自增，如果已是最后一个位置，重新设置 putIndex = 0;if (++putIndex == items.length)
        putIndex = 0;
    count++;//队列中元素数量加1
    //唤醒调用take()方法的线程，执行元素获取操作。    notEmpty.signal();
}

//put方法，阻塞时可中断publicvoid put(E e) throws InterruptedException {
     checkNotNull(e);
      final ReentrantLock lock = this.lock;
      lock.lockInterruptibly();//该方法可中断try {
          //当队列元素个数与数组长度相等时，无法添加元素while (count == items.length)
              //将当前调用线程挂起，添加到notFull条件队列中等待唤醒              notFull.await();
          enqueue(e);//如果队列没有满直接添加。。
      } finally {
          lock.unlock();
      }
  }

public E poll() {
      final ReentrantLock lock = this.lock;
       lock.lock();
       try {
           //判断队列是否为null，不为null执行dequeue()方法，否则返回nullreturn (count == 0) ? null : dequeue();
       } finally {
           lock.unlock();
       }
    }
 //删除队列头元素并返回private E dequeue() {
     //拿到当前数组的数据final Object[] items = this.items;
      @SuppressWarnings("unchecked")
      //获取要删除的对象
      E x = (E) items[takeIndex];
      将数组中takeIndex索引位置设置为null
      items[takeIndex] = null;
      //takeIndex索引加1并判断是否与数组长度相等，
      //如果相等说明已到尽头，恢复为0if (++takeIndex == items.length)
          takeIndex = 0;
      count--;//队列个数减1if (itrs != null)
          itrs.elementDequeued();//同时更新迭代器中的元素数据
      //删除了元素说明队列有空位，唤醒notFull条件对象添加线程，执行添加操作      notFull.signal();
      return x;
    }

publicboolean remove(Object o) {
    if (o == null) returnfalse;
    //获取数组数据final Object[] items = this.items;
    final ReentrantLock lock = this.lock;
    lock.lock();//加锁try {
        //如果此时队列不为null，这里是为了防止并发情况if (count > 0) {
            //获取下一个要添加元素时的索引finalint putIndex = this.putIndex;
            //获取当前要被删除元素的索引int i = takeIndex;
            //执行循环查找要删除的元素do {
                //找到要删除的元素if (o.equals(items[i])) {
                    removeAt(i);//执行删除returntrue;//删除成功返回true                }
                //当前删除索引执行加1后判断是否与数组长度相等
                //若为true，说明索引已到数组尽头，将i设置为0if (++i == items.length)
                    i = 0; 
            } while (i != putIndex);//继承查找        }
        returnfalse;
    } finally {
        lock.unlock();
    }
}
//根据索引删除元素，实际上是把删除索引之后的元素往前移动一个位置void removeAt(finalint removeIndex) {

     final Object[] items = this.items;
      //先判断要删除的元素是否为当前队列头元素if (removeIndex == takeIndex) {
          //如果是直接删除
          items[takeIndex] = null;
          //当前队列头元素加1并判断是否与数组长度相等，若为true设置为0if (++takeIndex == items.length)
              takeIndex = 0;
          count--;//队列元素减1if (itrs != null)
              itrs.elementDequeued();//更新迭代器中的数据
      } else {
      //如果要删除的元素不在队列头部，
      //那么只需循环迭代把删除元素后面的所有元素往前移动一个位置
          //获取下一个要被添加的元素的索引，作为循环判断结束条件finalint putIndex = this.putIndex;
          //执行循环for (int i = removeIndex;;) {
              //获取要删除节点索引的下一个索引int next = i + 1;
              //判断是否已为数组长度，如果是从数组头部（索引为0）开始找if (next == items.length)
                  next = 0;
               //如果查找的索引不等于要添加元素的索引，说明元素可以再移动if (next != putIndex) {
                  items[i] = items[next];//把后一个元素前移覆盖要删除的元
                  i = next;
              } else {
              //在removeIndex索引之后的元素都往前移动完毕后清空最后一个元素
                  items[i] = null;
                  this.putIndex = i;
                  break;//结束循环              }
          }
          count--;//队列元素减1if (itrs != null)
              itrs.removedAt(removeIndex);//更新迭代器数据      }
      notFull.signal();//唤醒添加线程    }
//从队列头部删除，队列没有元素就阻塞，可中断public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
      lock.lockInterruptibly();//中断try {
          //如果队列没有元素while (count == 0)
              //执行阻塞操作              notEmpty.await();
          return dequeue();//如果队列有元素执行删除操作
      } finally {
          lock.unlock();
      }
    }

}

            public
            class LinkedBlockingQueue<E> extends AbstractQueue<E>
        implements BlockingQueue<E>, java.io.Serializable {

    /**
     * 节点类，用于存储数据
     */staticclass Node<E> {
        E item;

        /**
         * One of:
         * - the real successor Node
         * - this Node, meaning the successor is head.next
         * - null, meaning there is no successor (this is the last node)
         */
        Node<E> next;

        Node(E x) { item = x; }
    }

    /** 阻塞队列的大小，默认为Integer.MAX_VALUE */privatefinalint capacity;

    /** 当前阻塞队列中的元素个数 */privatefinal AtomicInteger count = new AtomicInteger();

    /**
     * 阻塞队列的头结点
     */transient Node<E> head;

    /**
     * 阻塞队列的尾节点
     */privatetransient Node<E> last;

    /** 获取并移除元素时使用的锁，如take, poll, etc */privatefinal ReentrantLock takeLock = new ReentrantLock();

    /** notEmpty条件对象，当队列没有数据时用于挂起执行删除的线程 */privatefinal Condition notEmpty = takeLock.newCondition();

    /** 添加元素时使用的锁如 put, offer, etc */privatefinal ReentrantLock putLock = new ReentrantLock();

    /** notFull条件对象，当队列数据已满时用于挂起执行添加的线程 */privatefinal Condition notFull = putLock.newCondition();

publicboolean offer(E e) {
     //添加元素为null直接抛出异常if (e == null) thrownew NullPointerException();
      //获取队列的个数final AtomicInteger count = this.count;
      //判断队列是否已满if (count.get() == capacity)
          returnfalse;
      int c = -1;
      //构建节点
      Node<E> node = new Node<E>(e);
      final ReentrantLock putLock = this.putLock;
      putLock.lock();
      try {
          //再次判断队列是否已满，考虑并发情况if (count.get() < capacity) {
              enqueue(node);//添加元素
              c = count.getAndIncrement();//拿到当前未添加新元素时的队列长度
              //如果容量还没满if (c + 1 < capacity)
                  notFull.signal();//唤醒下一个添加线程，执行添加操作          }
      } finally {
          putLock.unlock();
      }
      // 由于存在添加锁和消费锁，而消费锁和添加锁都会持续唤醒等到线程，因此count肯定会变化。
      //这里的if条件表示如果队列中还有1条数据if (c == 0) 
        signalNotEmpty();//如果还存在数据那么就唤醒消费锁return c >= 0; // 添加成功返回true，否则返回false  }

//入队操作privatevoid enqueue(Node<E> node) {
     //队列尾节点指向新的node节点
     last = last.next = node;
}

//signalNotEmpty方法privatevoid signalNotEmpty() {
      final ReentrantLock takeLock = this.takeLock;
      takeLock.lock();
          //唤醒获取并删除元素的线程          notEmpty.signal();
      } finally {
          takeLock.unlock();
      }
  }
publicboolean remove(Object o) {
   if (o == null) returnfalse;
     fullyLock();//同时对putLock和takeLock加锁try {
         //循环查找要删除的元素for (Node<E> trail = head, p = trail.next;
              p != null;
              trail = p, p = p.next) {
             if (o.equals(p.item)) {//找到要删除的节点
                 unlink(p, trail);//直接删除returntrue;
             }
         }
         returnfalse;
     } finally {
         fullyUnlock();//解锁     }
    }

//两个同时加锁void fullyLock() {
       putLock.lock();
       takeLock.lock();
   }

void fullyUnlock() {
      takeLock.unlock();
      putLock.unlock();
  }
public E poll() {
         //获取当前队列的大小final AtomicInteger count = this.count;
        if (count.get() == 0)//如果没有元素直接返回nullreturnnull;
        E x = null;
        int c = -1;
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lock();
        try {
            //判断队列是否有数据if (count.get() > 0) {
                //如果有，直接删除并获取该元素值
                x = dequeue();
                //当前队列大小减一
                c = count.getAndDecrement();
                //如果队列未空，继续唤醒等待在条件对象notEmpty上的消费线程if (c > 1)
                    notEmpty.signal();
            }
        } finally {
            takeLock.unlock();
        }
        //判断c是否等于capacity，这是因为如果满说明NotFull条件对象上
        //可能存在等待的添加线程if (c == capacity)
            signalNotFull();
        return x;
    }

  private E dequeue() {
        Node<E> h = head;//获取头结点
        Node<E> first = h.next; 获取头结的下一个节点（要删除的节点）
        h.next = h; // help GC//自己next指向自己，即被删除
        head = first;//更新头结点
        E x = first.item;//获取删除节点的值
        first.item = null;//清空数据，因为first变成头结点是不能带数据的，这样也就删除队列的带数据的第一个节点return x;
    }
public E take() throws InterruptedException {
        E x;
        int c = -1;
        //获取当前队列大小final AtomicInteger count = this.count;
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lockInterruptibly();//可中断try {
            //如果队列没有数据，挂机当前线程到条件对象的等待队列中while (count.get() == 0) {
                notEmpty.await();
            }
            //如果存在数据直接删除并返回该数据
            x = dequeue();
            c = count.getAndDecrement();//队列大小减1if (c > 1)
                notEmpty.signal();//还有数据就唤醒后续的消费线程
        } finally {
            takeLock.unlock();
        }
        //满足条件，唤醒条件对象上等待队列中的添加线程if (c == capacity)
            signalNotFull();
        return x;
    }
}