ArrayBlockingQueue

一.使用

public class Test {
    public static void main(String[] args) {
        ArrayBlockingQueue<Integer> arrayBlockingQueue = new ArrayBlockingQueue<>(10);
        PutThread putThread = new PutThread(arrayBlockingQueue);
        GetThread getThread = new GetThread(arrayBlockingQueue);
        putThread.start();
        getThread.start();
    }
}
class PutThread extends Thread {
    private ArrayBlockingQueue<Integer> arrayBlockingQueue;
    public PutThread(ArrayBlockingQueue<Integer> arrayBlockingQueue) {
        this.arrayBlockingQueue = arrayBlockingQueue;
    }
    @SneakyThrows
    @Override
    public void run() {
        for (int i = 0; i < 10; i++) {
            System.out.println("put" + i);
            arrayBlockingQueue.put(i);
            Thread.sleep(50);
        }
    }
}
class GetThread extends Thread {
    private ArrayBlockingQueue<Integer> arrayBlockingQueue;
    public GetThread(ArrayBlockingQueue<Integer> arrayBlockingQueue) {
        this.arrayBlockingQueue = arrayBlockingQueue;
    }
    @SneakyThrows
    @Override
    public void run() {
        for (int i = 0; i < 15; i++) {
            System.out.println("take" + arrayBlockingQueue.take());
            Thread.sleep(50);
        }
    }
}
输出：
put0
take0
put1
take1
put2
take2
put3
take3
put4
take4
put5
take5
put6
take6
put7
take7
put8
take8
put9
take9

二.源码

构造方法

public ArrayBlockingQueue(int capacity) {
    this(capacity, false);
}
public ArrayBlockingQueue(int capacity, boolean fair) {
    // final Object[] items;
    this.items = new Object[capacity];
    lock = new ReentrantLock(fair);
    notEmpty = lock.newCondition();
    notFull =  lock.newCondition();
}

put

public void put(E e) throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == items.length)
            notFull.await();
        enqueue(e);
    } finally {
        lock.unlock();
    }
}
private void enqueue(E x) {
    final Object[] items = this.items;
    items[putIndex] = x;
    if (++putIndex == items.length)
        putIndex = 0;
    count++;
    notEmpty.signal();
}

take

public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
    lock.lockInterruptibly();
    try {
        while (count == 0)
            notEmpty.await();
        return dequeue();
    } finally {
        lock.unlock();
    }
}
private E dequeue() {
    final Object[] items = this.items;
    E x = (E) items[takeIndex];
    items[takeIndex] = null;
    if (++takeIndex == items.length)
        takeIndex = 0;
    count--;
    notFull.signal();
    return x;
}