并行设计模式--生产者消费者

// 队列本身
final Object[] items;
// 头指针
int takeIndex;
// 位指针
int putIndex;
// 队列内元素数量,也因此此队列有界,虽然int最大有21亿多.
int count;

// 重入锁
final ReentrantLock lock;
// 用于控制消费者线程的条件
private final Condition notEmpty;
// 用于控制生产者线程的条件
private final Condition notFull;

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    // 操作1： 获取独占锁
    lock.lockInterruptibly();
    try {
        // 操作2：队列为空则主动await
        while (count == 0)
            notEmpty.await();
        // 操作3：出队，见下面的代码详情
        return dequeue();
    } finally {
        lock.unlock();
    }
}

private E dequeue() {
    // 操作3：出队操作实际上是把数组中对应元素返回，然后数组置为NULL
    final Object[] items = this.items;
    @SuppressWarnings("unchecked")
    E x = (E) items[takeIndex];
    items[takeIndex] = null;
    if (++takeIndex == items.length)
        takeIndex = 0;
    count--;
    if (itrs != null)
        itrs.elementDequeued();
    // 操作3：唤醒生产者线程
    notFull.signal();
    return x;
}

public void put(E e) throws InterruptedException {
    checkNotNull(e);
    final ReentrantLock lock = this.lock;
    // 操作1：获取独占锁
    lock.lockInterruptibly();
    try {
        //操作2：队列已满主动休眠
        while (count == items.length)
            notFull.await();
        // 操作3：入队操作，见下方详细分析
        enqueue(e);
    } finally {
        lock.unlock();
    }
}

private void enqueue(E x) {
    // 操作3：入队实际上是往数组中添加一个元素，因为是固定容量数组，扩容之类的都不需要考虑
    final Object[] items = this.items;
    items[putIndex] = x;
    if (++putIndex == items.length)
        putIndex = 0;
    count++;
    // 操作3：唤醒消费者线程
    notEmpty.signal();
}

 // 该单链表最大长度
 private final int capacity;
 // 该单链表实际长度,不同于数组可以根据下标计算,因此这里需要额外记录
 private final AtomicInteger count = new AtomicInteger();
 // 头指针,其item为null
transient Node<E> head;
 // 尾指针，其next为null
private transient Node<E> last;
 // 读操作锁
private final ReentrantLock takeLock = new ReentrantLock();
 // 用于消费者线程操作
private final Condition notEmpty = takeLock.newCondition();
 // 写操作锁
private final ReentrantLock putLock = new ReentrantLock();
 // 用于生产者线程操作
private final Condition notFull = putLock.newCondition();

public E take() throws InterruptedException {
    E x;
    int c = -1; // 本次操作标识
    final AtomicInteger count = this.count;
    final ReentrantLock takeLock = this.takeLock;
    // 操作1：获取独占锁
    takeLock.lockInterruptibly();
    try {
        // 操作2：队列中无元素，主动休眠
        while (count.get() == 0) {
            notEmpty.await();
        }
        // 操作3：出队，出队是链表节点的指针引用切换，就不贴代码了
        x = dequeue();
        
        c = count.getAndDecrement();
        // 操作4：队列中还有元素则唤醒其他消费者再次消费
        if (c > 1)
            notEmpty.signal();
    } finally {
        takeLock.unlock();
    }
    // 附加操作：c为出队前的队列长度，当c等于队列容量说明之前队列是满的状态，那么所有的生产者都可能休眠中，因此这里需要唤醒。
    if (c == capacity)
        signalNotFull();
    return x;
}

public void put(E e) throws InterruptedException {
    if (e == null) throw new NullPointerException();
    int c = -1;
    Node<E> node = new Node<E>(e);
    final ReentrantLock putLock = this.putLock;
    final AtomicInteger count = this.count;
    // 操作1： 获取独占锁
    putLock.lockInterruptibly();
    try {
        // 操作2：当队列已满，则生产者主动await
        while (count.get() == capacity) {
            notFull.await();
        }
        // 操作3：出队操作
        enqueue(node);

        c = count.getAndIncrement();
        // 操作4：出队后队列仍有容量，则唤醒其他生产者进行生产
        if (c + 1 < capacity)
            notFull.signal();
    } finally {
        putLock.unlock();
    }
    // 附加操作：当c为0时，也就是入队之前队列为0，此时消费者有可能都在await状态，因此入队之后需要唤醒对应的消费者进行消费。
    if (c == 0)
        signalNotEmpty();
}