BlockingQueue介绍

• ArrayBlockingQueue： 一个由数组结构组成的有界阻塞队列。
• LinkedBlockingQueue： 一个由链表结构组成的有界阻塞队列。
• SynchronousQueue： 一个不存储元素的阻塞队列。
• PriorityBlockingQueue： 一个支持优先级排序的无界阻塞队列。
  插入方法 add offer put offer
  删除方法 remove poll take poll

BlockingQueue用法

带着疑问去使用，可能会更好：

• 队列空着的时候去拿数据会怎么样？
• 队列满着的时候去存数据会怎么样？
• 前面的方法，能否会出现阻塞或者非阻塞现象？

队列满了，存数据会怎么样？

队列满了，那么调用put方法，会将当前线程阻塞。

public static void main(String[] args) throws InterruptedException {
    ArrayBlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(10);
 
    for (int i = 0 ; i < 15 ; i++) {
        final int i1 = i;
        Thread t1 = new Thread(() -> {
            try {
                blockingQueue.put(" msg " + i1);
                System.out.println("======================= provider " + i1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
 
        t1.start();
    }
 
    Thread t2 = new Thread(() -> {
        try {
            String msg = blockingQueue.take();
            System.out.println("======================= consumer " + msg);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    });
 
    t2.start();
}

队列空了，取数据会怎么样？

队列空了，那么调用take方法，会将当前线程阻塞。

public static void main(String[] args) throws InterruptedException {
    ArrayBlockingQueue<String> blockingQueue = new ArrayBlockingQueue<>(10);
 
    for (int i = 0 ; i < 5 ; i++) {
        final int i1 = i;
 
        Thread t1 = new Thread(() -> {
            try {
                blockingQueue.put(" msg " + i1);
                System.out.println("======================= provider " + i1);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
 
        t1.start();
    }
 
    for (int j = 0 ; j < 10 ; j++) {
        Thread t2 = new Thread(() -> {
            try {
                String msg = blockingQueue.take();
                System.out.println("======================= consumer " + msg);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
 
        t2.start();
    }
}

poll方法，在队列空了的时候，返回false，取出的对象为空，当前线程非阻塞。
offer方法，在队列满了的时候，返回false，当前线程非阻塞。

源码

ArrayBlockingQueue构造器初始化参数

public ArrayBlockingQueue(int capacity) {
    this(capacity, false);
}
    
public ArrayBlockingQueue(int capacity, boolean fair) {
    // 容量参数小于0，报非法参数异常
    if (capacity <= 0)
        throw new IllegalArgumentException();
    // 初始化存入元素的数组
    this.items = new Object[capacity];
    // 初始化非公平的重入锁
    lock = new ReentrantLock(fair);
    // 设置非空的等待唤醒队列
    notEmpty = lock.newCondition();
    // 设置非满的等待唤醒队列
    notFull =  lock.newCondition();
}

put阻塞存入数据方法

public void put(E var1) throws InterruptedException {
    checkNotNull(var1);
    // 重入锁上锁
    ReentrantLock var2 = this.lock;
    var2.lockInterruptibly();
 
    try {
        // 当队列已经满了，那么就将当前线程等待
        while(this.count == this.items.length) {
            this.notFull.await();
        }
        // 进队列
        this.enqueue(var1);
    } finally {
        // 重入锁解锁
        var2.unlock();
    }
}
 
private void enqueue(E var1) {
    // 获取存放对象的数组
    Object[] var2 = this.items;
    // 将元素存入数组
    var2[this.putIndex] = var1;
    // 如果当前的下标已经超出了存放对象数组的长度
    if (++this.putIndex == var2.length) {
        this.putIndex = 0;
    }
    
    ++this.count;
    // 唤醒非空等待队列中的元素
    this.notEmpty.signal();
}

take阻塞拿取数据方法

public E take() throws InterruptedException {
    ReentrantLock var1 = this.lock;
    var1.lockInterruptibly();
 
    Object var2;
    try {
        // 当前元素的个数已经为0
        while(this.count == 0) {
            // 当前线程等待
            this.notEmpty.await();
        }
        // 出队列
        var2 = this.dequeue();
    } finally {
        var1.unlock();
    }
 
    return var2;
}
 
private E dequeue() {
    // 获取存放元素的数组
    Object[] var1 = this.items;
    // 通过记录的取出下标索引取出元素
    Object var2 = var1[this.takeIndex];
 
    var1[this.takeIndex] = null;
    // 存放元素的数组已经到达的最后，没有元素了
    if (++this.takeIndex == var1.length) {
        this.takeIndex = 0;
    }
    // 记录元素的个数减1
    --this.count;
    if (this.itrs != null) {
        this.itrs.elementDequeued();
    }
    // 唤醒所有的未满的等待队列中的元素
    this.notFull.signal();
    return var2;
}