阻塞队列与普通的队列(LinkedList/ArrayList)相比,支持在向队列中添加元素时,队列的长度已满阻塞当前添加线程,直到队列未满或者等待超时;从队列中获取元素时,队列中元素为空 ,会将获取元素的线程阻塞,直到队列中存在元素 或者等待超时。
在JUC包中常用的阻塞队列包含ArrayBlockingQueue/LinkedBlockingQueue/LinkedBlockingDeque等,从结构来看都继承了AbstractQueue实现了BlockingQueue接口(LinkedBlockingDeque是双向阻塞队列,实现的是BlockingDeque接口),在BlockingQueue接口中定义了几个供子类实现的接口,可以分为3部分,puts操作、takes操作、其他操作。
puts操作 add(E e) : 添加成功返回true,失败抛IllegalStateException异常 offer(E e) : 成功返回 true,如果此队列已满,则返回 false(如果添加了时间参数,且队列已满也会阻塞) put(E e) :将元素插入此队列的尾部,如果该队列已满,则一直阻塞 takes操作 remove(Object o) :移除指定元素,成功返回true,失败返回false poll() : 获取并移除此队列的头元素,若队列为空,则返回 null(如果添加了时间参数,且队列中没有数据也会阻塞) take():获取并移除此队列头元素,若没有元素则一直阻塞。 peek() :获取但不移除此队列的头;若队列为空,则返回 null。 other操作 contains(Object o):队列中是否包含指定元素 drainTo(Collection<? super E> c):队列转化为集合
关于阻塞队列,我们主要看LinkedBlockingQueue与ArrayBlockingQueue.
ArrayBlockingQueue
ArrayBlockingQueue是基于数组的、有界的、遵循FIFO原则的阻塞队列,队列初始化时必须指定队列的长度。
这是一个经典的“有界缓冲区”,其中固定大小的数组包含由生产者插入并由消费者提取的元素。创建后,无法更改容量。此类支持用于排序等待生产者和消费者线程的可选公平策略。默认情况下,不保证此顺序。但是,将fairness设置为true构造的队列以FIFO顺序授予线程访问权限。公平性通常会降低吞吐量,但会降低可变性并避免饥饿。
结构
相关变量
final Object[] items; //一个数组,用来存放队列中的变量(队列的基础) int count; //队列中元素的数量 int takeIndex; //下一次take、poll、remove、peek操作的下标值 int putIndex; //下次add、offer、put操作的下标值
构造函数
ArrayBlockingQueue提供了三个构造函数,在只传递初始化大小值时,默认使用的锁是非公平锁,对比三个不同的构造函数而言,真正初始化队列的构造方法是ArrayBlockingQueue(int capacity, boolean fair)方法,传入集合的构造方法会在调用该方法后将集合遍历存入队列中
public ArrayBlockingQueue(int capacity) { this(capacity, false); } public ArrayBlockingQueue(int capacity, boolean fair) { if (capacity <= 0) throw new IllegalArgumentException(); this.items = new Object[capacity]; lock = new ReentrantLock(fair); //使用同一个锁对象,此处是与LinkedBlockingQueue(使用两个不同的锁来控制添加,取出操作)不同的地方 notEmpty = lock.newCondition(); notFull = lock.newCondition(); } public ArrayBlockingQueue(int capacity, boolean fair, Collection<? extends E> c) { this(capacity, fair); final ReentrantLock lock = this.lock; lock.lock(); // Lock only for visibility, not mutual exclusion try { int i = 0; try { for (E e : c) { checkNotNull(e); //向数组中添加数据 items[i++] = e; } } catch (ArrayIndexOutOfBoundsException ex) { throw new IllegalArgumentException(); } count = i; //设置下次添加操作对应的数组下标值 putIndex = (i == capacity) ? 0 : i; } finally { lock.unlock(); } }
offer/add操作
add本质上调用的是offer操作,通过返回值true/false可以判断队列中添加元素是否成功,队列已满返回false
public boolean offer(E e) { checkNotNull(e); final ReentrantLock lock = this.lock; lock.lock(); try { //队列已满,直接返回false if (count == items.length) return false; else { enqueue(e); return true; } } finally { lock.unlock(); } } //入队列操作 private void enqueue(E x) { // assert lock.getHoldCount() == 1; // assert items[putIndex] == null; final Object[] items = this.items; items[putIndex] = x; if (++putIndex == items.length) putIndex = 0; count++; //唤醒take/poll(有时间参数)方法获取数据的线程 notEmpty.signal(); }
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(); } }
poll操作
队列为空返回null,对于有时间参数的poll操作,在队列为空时,会被挂起等待
public E poll() { final ReentrantLock lock = this.lock; lock.lock(); try { return (count == 0) ? null : dequeue(); } finally { lock.unlock(); } } private E dequeue() { // assert lock.getHoldCount() == 1; // assert items[takeIndex] != 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(); //唤醒puts线程 notFull.signal(); return x; }
take操作
队列为空等待,直到队列中存在元素被唤醒
public E take() throws InterruptedException { final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try { while (count == 0) notEmpty.await(); return dequeue(); } finally { lock.unlock(); } }
peek操作
获取队列中第一个不为空的元素(每次takes操作,或者puts操作都会设置下次takes/puts操作的下标)
public E peek() { final ReentrantLock lock = this.lock; lock.lock(); try { //通过下标值获取元素 return itemAt(takeIndex); // null when queue is empty } finally { lock.unlock(); } } final E itemAt(int i) { return (E) items[i]; }
remove操作
删除内部元素操作是一种本质上缓慢且具有破坏性的操作,需要将删除元素后的元素统一迁移一个单位,并且在操作过程中会获得锁,对性能有影响,因此不应轻易执行remove操作
public boolean remove(Object o) { if (o == null) return false; final Object[] items = this.items; final ReentrantLock lock = this.lock; lock.lock(); try { if (count > 0) { final int putIndex = this.putIndex; int i = takeIndex; //遍历队列 do { if (o.equals(items[i])) { removeAt(i); return true; } if (++i == items.length) i = 0; } while (i != putIndex); } return false; } finally { lock.unlock(); } } void removeAt(final int removeIndex) { // assert lock.getHoldCount() == 1; // assert items[removeIndex] != null; // assert removeIndex >= 0 && removeIndex < items.length; final Object[] items = this.items; //如果需要移除的元素下标值为下一次取数的下标值,执行类似取数的操作 if (removeIndex == takeIndex) { // removing front item; just advance items[takeIndex] = null; //如果takes下标到达队列最大长度,归零 if (++takeIndex == items.length) takeIndex = 0; count--; if (itrs != null) itrs.elementDequeued(); } else { // an "interior" remove // slide over all others up through putIndex. final int putIndex = this.putIndex; //如果不相等,将需删除元素的后续元素统一迁移一位 for (int i = removeIndex;;) { int next = i + 1; if (next == items.length) next = 0; if (next != putIndex) { items[i] = items[next]; i = next; } else { //移动完成,设置puts操作下标 items[i] = null; this.putIndex = i; break; } } count--; if (itrs != null) itrs.removedAt(removeIndex); } //唤醒put操作等待的线程 notFull.signal(); }
综上,ArrayBlockingQueue队列的逻辑并不复杂,但需要注意一下几点
1.ArrayBlockingQueue是以数组来实现队列功能的,在执行puts或者takes操作时一旦下一个操作的下标值大于队列的长度,类中用来记录存取下标值会归零,已达到循环使用队列的目的 2.ArrayBlockingQueue是通过一个锁控制takes以及puts操作,说明在同一时间内只能执行takes操作或者puts操作中的一种,对于阻塞队列来说,保证了线程安全,
但是会影响队列的消费和生产效率,并发性会下降 3.ArrayBlockingQueue在执行remove操作时,会将整个数组进行移动(最坏情况下),同时还会获得锁,对性能的影响比较大
LinkedBlockingQueue
LinkedBlockingQueue是基于链表的、有界的、遵循FIFO原则的阻塞队列,队列默认的最大长度为Integer.MAX_VALUE
结构
重要属性
private final int capacity;//队列的大小,可以自定义,默认为Integer.MAX_VALUE
private final AtomicInteger count = new AtomicInteger(); //当前队列中元素的数量
//take、poll操作需要持有的锁,LinkedBlockingQueue支持并发操作,对于从队列中获取数据需要加锁(会阻塞,ConcurrentLinkedQueue/ConcurrentLinkedDeque
是使用CAS操作来控制的,不会出现阻塞的问题) private final ReentrantLock takeLock = new ReentrantLock();
//put、offer操作需要持有的锁,同上 private final ReentrantLock putLock = new ReentrantLock();
//notEmpty条件对象,当队列没有数据时用于挂起执行删除的线程 private final Condition notEmpty = takeLock.newCondition();
//notFull条件对象,当队列数据已满时用于挂起执行添加的线程 private final Condition notFull = putLock.newCondition();
Node类
相对于其他(ConcurrentLinkedQueue/ConcurrentLinkedDeque)类来说,LinkedBlockingQueue的Node类要简单好多,由于是基于单链表实现的,只有一个next属性(保存后继节点),一个item属性(存放值),一个构造函数
static class Node<E> { E item; Node<E> next; Node(E x) { item = x; } }
构造函数
LinkedBlockingQueue提供了三个构造函数,在不传参数的情况下,默认队列的大小为Integer.MAX_VALUE,对比三个不同的构造函数而言,真正初始化 队列的构造方法是LinkedBlockingQueue(int capacity)方法,传入集合的构造方法会在调用该方法后将集合遍历存入队列中
public LinkedBlockingQueue() { this(Integer.MAX_VALUE); } public LinkedBlockingQueue(int capacity) { if (capacity <= 0) throw new IllegalArgumentException(); //设置队列大小,new一个null节点,head、tail节点指向改节点 this.capacity = capacity; last = head = new Node<E>(null); } public LinkedBlockingQueue(Collection<? extends E> c) { this(Integer.MAX_VALUE); //获取put、offer操作需要的锁,可重入 final ReentrantLock putLock = this.putLock; putLock.lock(); // Never contended, but necessary for visibility try { int n = 0; for (E e : c) { if (e == null) throw new NullPointerException(); if (n == capacity) throw new IllegalStateException("Queue full"); //将队列的last节点指向该节点 enqueue(new Node<E>(e)); ++n; } //队列元素计数 count.set(n); } finally { putLock.unlock(); } }
offer操作
通过返回值true/false判断是否成功,offer操作当队列满后并不会阻塞(有时间参数的offer操作也会阻塞),而是直接返回false,put操作是没有返回值的,并且会一直阻塞,等待被唤醒(或者超过时间抛出异常)
public boolean offer(E e) { if (e == null) throw new NullPointerException(); final AtomicInteger count = this.count; //是否超过最大值 if (count.get() == capacity) return false; int c = -1; Node<E> node = new Node<E>(e); final ReentrantLock putLock = this.putLock; //获取锁 putLock.lock(); try { //再次判断队列是否存放满(可能存在多线程的情况) if (count.get() < capacity) { enqueue(node); c = count.getAndIncrement(); //如果队列没有放满,唤醒下一个添加线程 if (c + 1 < capacity) //其实这个地方,唤醒的添加线程是执行put方法(或者offer有时间参数的操作)时被阻塞的线程,如果仅仅只是执行offer操作应该不会执行任何操作, 没有对应的添加线程添加到条件队列中(个人理解,也是不太理解的地方) notFull.signal(); } } finally { putLock.unlock(); } if (c == 0) //当c=0时,表明当前队列中存在一个元素,通知消费线程去消费 signalNotEmpty(); return c >= 0; } private void signalNotEmpty() { final ReentrantLock takeLock = this.takeLock; takeLock.lock(); try { //唤醒条件等待队列中的消费者去消费数据 notEmpty.signal(); } finally { takeLock.unlock(); } }
put操作
put方法不会像offer方法那样去检查队列大小是否超过设定值,put操作一个元素入队列时,如果队列已满,当前线程会进入nofull的条件等待队列中等待,直到队列中元素个数小于队列大小时被唤醒,才继续put操作
public void put(E e) throws InterruptedException { if (e == null) throw new NullPointerException(); // Note: convention in all put/take/etc is to preset local var // holding count negative to indicate failure unless set. int c = -1; Node<E> node = new Node<E>(e); final ReentrantLock putLock = this.putLock; final AtomicInteger count = this.count; putLock.lockInterruptibly(); try { /* * Note that count is used in wait guard even though it is * not protected by lock. This works because count can * only decrease at this point (all other puts are shut * out by lock), and we (or some other waiting put) are * signalled if it ever changes from capacity. Similarly * for all other uses of count in other wait guards. */ //队列已满存放线程阻塞,等待被唤醒 while (count.get() == capacity) { notFull.await(); } //入队列 enqueue(node); c = count.getAndIncrement(); //队列没满,唤醒notfull等待队列中的添加线程 if (c + 1 < capacity) notFull.signal(); } finally { putLock.unlock(); } if (c == 0) //当c=0时,表明当前队列中存在一个元素,通知消费线程去消费 signalNotEmpty(); }
poll操作
必定会有返回值(异常除外),但包含null(队列中没有数据),与take操作比较可以发现,take操作在队列中没有数据时执行take操作的线程会被挂起,直到队列中有数据(有时间参数的poll操作也会被挂起,等待唤醒或者超时)
public E poll() { final AtomicInteger count = this.count; //如果队列没有数据,返回null if (count.get() == 0) return null; E x = null; int c = -1; final ReentrantLock takeLock = this.takeLock; takeLock.lock(); try { if (count.get() > 0) { x = dequeue(); c = count.getAndDecrement(); if (c > 1) //唤醒其他消费线程 notEmpty.signal(); } } finally { takeLock.unlock(); } if (c == capacity) signalNotFull(); return x; } //出队列操作,因为队列的head节点为null节点,在出队列时,会始终保持head节点为空,next节点为真正意义上的首节点 private E dequeue() { // assert takeLock.isHeldByCurrentThread(); // assert head.item == null; Node<E> h = head; Node<E> first = h.next; //自指向,该节点已经无用,便于GC h.next = h; // help GC head = first; E x = first.item; first.item = null; return x; }
take操作
take操作不会向poll操作去检查队列中有没有数据,队列中没有数据时会被挂起,等待被唤醒
public E take() throws InterruptedException { E x; int c = -1; final AtomicInteger count = this.count; final ReentrantLock takeLock = this.takeLock; takeLock.lockInterruptibly(); try { //队列中是否有数据,没有等待 while (count.get() == 0) { notEmpty.await(); } x = dequeue(); c = count.getAndDecrement(); if (c > 1) //如果队列中有数据存在,唤醒notempty等待队列中的消费线程 notEmpty.signal(); } finally { takeLock.unlock(); } if (c == capacity) //唤醒添加线程 signalNotFull(); return x; }
peek操作
获取队列中头部元素,可能存在其他线程执行的删除、take(poll)操作,所以要加锁,获取数据不一定准确
public E peek() { if (count.get() == 0) return null; final ReentrantLock takeLock = this.takeLock; takeLock.lock(); try { //head节点不存放数据,所以取的是next节点 Node<E> first = head.next; if (first == null) return null; else return first.item; } finally { takeLock.unlock(); } }
remove操作
在执行删除操作时,会将puts以及takes操作都上锁,保证线程安全,然后执行遍历删除操作,在删除后,会去唤醒等待中的添加线程执行 添加操作
public boolean remove(Object o) { if (o == null) return false; fullyLock(); try { //遍历单项链表 for (Node<E> trail = head, p = trail.next; p != null; trail = p, p = p.next) { if (o.equals(p.item)) { //移除数据 unlink(p, trail); return true; } } return false; } finally { fullyUnlock(); } } void unlink(Node<E> p, Node<E> trail) { // assert isFullyLocked(); // p.next is not changed, to allow iterators that are // traversing p to maintain their weak-consistency guarantee. p.item = null; trail.next = p.next; if (last == p) last = trail; //唤醒添加线程 if (count.getAndDecrement() == capacity) notFull.signal(); }
综上,LinkedBlockingQueue的api与ArrayBlockingQueue并无太大差别,在实现思想上,LinkedBlockingQueue使用了锁分离以及链表,其他与ArraBlockingQueue(一个锁统一管理、数组)没太大区别
LinkedBlockingQueue与ArrayBlockingQueue异同
1.LinkedBlockingQueue是基于链表实现的初始化是可以不用指定队列大小(默认是Integer.MAX_VALUE);ArrayBlockingQueue是基于数组实现的初始化时必须指定队列大小 2.LinkedBlockingQueue在puts操作都会生成新的Node对象,takes操作Node对象在某一时间会被GC,可能会影响GC性能;ArrayBlockingQueue是固定的数组长度循环使用,
不会出现对象的产生与回收 3.LinkedBlockingQueue是基于链表的形式,在执行remove操作时,不用移动其他数据;ArrayBlockingQueue是基于链表,在remove时需要移动数据,影响性能 4.LinkedBlockingQueue使用两个锁将puts操作与takes操作分开;ArrayBlockingQueue使用一个锁来控制,在高并发高吞吐的情况下,LinkedBlockingQueue的性能较好