Java并发-BlockingQueue

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();
}

public LinkedBlockingQueue(int capacity) {
  if (capacity <= 0) throw new IllegalArgumentException();
  // 指定队列大小
  this.capacity = capacity;
  last = head = new Node<E>(null);
}

// 单链表节点Node
static class Node<E> {
  E item;
  Node<E> next;
  Node(E x) { item = x; }
}

public PriorityBlockingQueue(int initialCapacity,
                             Comparator<? super E> comparator) {
  if (initialCapacity < 1)
    throw new IllegalArgumentException();
  this.lock = new ReentrantLock();  // 非公平锁
  this.notEmpty = lock.newCondition();
  this.comparator = comparator;
  this.queue = new Object[initialCapacity];
}


// offer方法部分代码
Comparator<? super E> cmp = comparator;
if (cmp == null)
  siftUpComparable(n, e, array);
else
  siftUpUsingComparator(n, e, array, cmp);

// fair为true，等待线程将以FIFO的顺序进行访问
public SynchronousQueue(boolean fair) {
  transferer = fair ? new TransferQueue<E>() : new TransferStack<E>();
}

/** The queued items */
// 数据元素数组
final Object[] items;

/** items index for next take, poll, peek or remove */
// 下一个待获取元素索引
int takeIndex;

/** items index for next put, offer, or add */
// 下一个待插入元素索引
int putIndex;

/** Number of elements in the queue */
// 队列中元素个数
int count;

/*
 * Concurrency control uses the classic two-condition algorithm
 * found in any textbook.
 */

/** Main lock guarding all access */
// 所有访问的主锁
final ReentrantLock lock;

/** Condition for waiting takes */
// 消费者监视器
private final Condition notEmpty;

/** Condition for waiting puts */
// 生产者监视器
private final Condition notFull;


// 
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();
}

// 在队列尾部插入元素，若队列已满则等待队列非满。
public void put(E e) throws InterruptedException {
  // 校验插入元素，为空则抛出NPE
  checkNotNull(e);
  final ReentrantLock lock = this.lock;
  // 1. 尝试获取锁(响应中断)
  lock.lockInterruptibly();
  try {
    // 2. 当队列满时
    while (count == items.length)
      // 2.1 若队列满，则阻塞当前线程。等待`notFull.signal()`唤醒。
      notFull.await();
    // 3. 非满则执行入队操作
    enqueue(e);
  } finally {
    lock.unlock();
  }
}

// 在`putIndex`处放置当前元素，只有获取lock锁后才会调用
private void enqueue(E x) {
  // assert lock.getHoldCount() == 1;
  // assert items[putIndex] == null;
  final Object[] items = this.items;
  // 在`putIndex`处放置元素
  items[putIndex] = x;
  // putIndex等于数组长度时，重置为0索引。
  if (++putIndex == items.length)
    putIndex = 0;
  // 数量加1
  count++;
  // 4. 唤醒一个等待线程(等待取元素的线程)
  notEmpty.signal();
}

// 从头部获取元素，若队列为空则等待队列非空。
public E take() throws InterruptedException {
  final ReentrantLock lock = this.lock;
  // 1. 获取锁
  lock.lockInterruptibly();
  try {
    // 2. 当队列为空时
    while (count == 0)
      // 2.1 当队列为空时，阻塞当前线程。等待`notEmpty.signal()`唤醒。
      notEmpty.await();
    // 3. 非空则进行入队操作
    return dequeue();
  } finally {
    lock.unlock();
  }
}

// 从`takeIndex`位置获取当前元素，只有获取到lock锁后才会调用
private E dequeue() {
  // assert lock.getHoldCount() == 1;
  // assert items[takeIndex] != null;
  final Object[] items = this.items;
  @SuppressWarnings("unchecked")
  // 从`takeIndex`位置获取元素，然后清除该位置元素
  E x = (E) items[takeIndex];
  items[takeIndex] = null;
  // 
  if (++takeIndex == items.length)
    takeIndex = 0;
  // 队列元素减1
  count--;
  if (itrs != null)
    itrs.elementDequeued();
  // 4. 唤醒一个标记为notFull(生产者)的线程
  notFull.signal();
  return x;
}