高并发编程-队列-BlockingQueue-ArrayBlockingQueue
在介绍ArrayBlockingQueue之前,为了让大家更好的理解这个知识点,先把Queue和BlockingQueue的相关知识做个简单的介绍
一、Queue队列接口
Queue继承于Collection数据集合,Queue内部主要方法有六个,下面依次对着六个方法做简单介绍
public interface Queue<E> extends Collection<E>{ //本方法为向队列中添加元素,如果队列中空间够用,元素插入队列返回true,如果空间不够用将会抛出 IllegalStateException异常 boolean add(E e); //向队列中添加元素,添加成功返回true,如果队列已满,返回false boolean offer(E e); //删除队首元素并删除,如果队列为空将会抛NoSuchElementException 异常 E remove(); // 返回并删除队首元素,如果队列为空则返回null E poll(); //返回首部元素但不删除,如果队列为空则抛出异常NoSuchElementException E element(); // 返回首部元素但不删除,如果队列为空则返回null E peek(); }
二、BlockingQueue接口
BlockingQueue叫做阻塞队列,这个队列继承自Queue,所以也就继承了Queue中的方法,同时又扩展出了自己的两个附加操作队列
public interface BlockingQueue<E> extends Queue<E> { /** *同上*/ boolean add(E e); /** *同上*/ boolean offer(E e); /** *队列不满时候正常插入,当队列数据满时,阻塞。直到队列不满时候再次插入数据,支持中断*/ void put(E e) throws InterruptedException; /** * 向队列中插入数据,设置阻塞时间,如果超过阻塞时间还没有插入数据则返回false*/ boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException; /** * 队列中有数据则获取数据并删除,如果队列中没有数据则阻塞等待,直到有数据再次获取数据*/ E take() throws InterruptedException; /** *获取队列中的数据,如果有数据则获取数据,如果没有数据则阻塞,超过阻塞时间则返回null*/ E poll(long timeout, TimeUnit unit) throws InterruptedException; /** * 同上*/ int poll(); /** * 同上*/ boolean remove(Object o); }
BlockingQueue 方法总结
如果队列数据满了或者空时,各方法返回的结果总结
三、ArrayBlockingQueue
ArrayBlockingQueue是有限空间的队列,又叫做有界队列,所以在初始化时需要制定容量大小。其底层所采用的是数组方式的数据存储,利用ReentrantLock所来保证它的线程安全。
使用场景:用于生产者和消费者的模式,当生产者和消费者的速度基本匹配时,选用ArrayBlockingQueue是比较合适的,如果消费者速度大于生产者,队列为空,消费者线程就会被阻塞,反之生产者就会被阻塞。
并发的处理:ArrayBlockingQueue采用的是ReentrantLock锁,入队出队时用的同一个所,这样产生的问题就是入队和出队时不能并发执行,特高并发下性能不是很好
ArrayBlockingQueue部分源码介绍
内部主要是一把ReentrantLock锁,产生的两个条件队列 notFull notFull
//数据元素 final Object[] items; //消费下一个元素 int takeIndex; //插入下一个元素 int putIndex; //元素的总数 int count; //定义全局的锁 final ReentrantLock lock; //当队列为空时,消费者需要等待 private final Condition notEmpty; //当队列满时,生产者需要等待 private final Condition notFull; //构造方法,fair如不传数据,默认是false public ArrayBlockingQueue(int capacity, boolean fair) { if (capacity <= 0) throw new IllegalArgumentException(); //定义数据大写 this.items = new Object[capacity]; //创建全局锁,false默认为非公平锁, true为公平锁 lock = new ReentrantLock(fair); //获取锁的条件队列 notEmpty = lock.newCondition(); notFull = lock.newCondition(); }
put 数据时的源码介绍
//阻塞队列插入 public void put(E e) throws InterruptedException { //检查元素是否为空,如果为空直接抛出异常 checkNotNull(e); //获取锁 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; @SuppressWarnings("unchecked") //获取takeindex位置的元素 E x = (E) items[takeIndex]; //把对应位置的数组制空 items[takeIndex] = null; //如果环形数组到了最后 以为,在次指向首位 if (++takeIndex == items.length) takeIndex = 0; //总数减去 count--; if (itrs != null) itrs.elementDequeued(); //唤醒生产线程 notFull.signal(); return x; }