LinkedBlockingQueue:
主要成员变量:
public class LinkedBlockingQueue<E> extends AbstractQueue<E> implements BlockingQueue<E>, java.io.Serializable { /** * 链表节点类 */ static class Node<E> { E item; //节点的后继 Node<E> next; Node(E x) { item = x; } } /** 阻塞队列的最大容量, 如果未设置的话，该值为：Integer.MAX_VALUE */ private final int capacity; /** 队列中当前的元素个数 */ private final AtomicInteger count = new AtomicInteger(); /** * 链表的头结点. * 头结点不存放具体元素：head.item == null */ transient Node<E> head; /** * 链表的尾节点. * 尾节点的后继为null: last.next == null */ private transient Node<E> last; /** 出队锁，take, poll操作时需要加该锁 */ private final ReentrantLock takeLock = new ReentrantLock(); /** 出队条件 */ private final Condition notEmpty = takeLock.newCondition(); /** 入队锁，put, offer操作时需要加该锁 */ private final ReentrantLock putLock = new ReentrantLock(); /** 入队条件 */ private final Condition notFull = putLock.newCondition(); }
LinkedBlockingQueue内部的具体实现是一个自定义链表，维护了链表的头结点和尾节点，从头结点出队，从尾节点入队，从而实现了队列的功能。
2.阻塞有两重含义：
（1）当队列满了的时候阻塞入队，也就是说如果有线程往队列里面添加元素，如果此时队列已满，那么该线程将被阻塞直到其他线程消费队列里的元素为止。
（2）当队列为空的时候阻塞出队，也就是说如果有线程从队列里面获取元素，如果此时队列为空，那么该线程将被阻塞直到其他线程往队列里面写入元素为止。
3.入队操作:
/** * 向队列尾部插入一个元素，如果没有空间则阻塞直到空间可用 * * @throws InterruptedException {@inheritDoc} * @throws NullPointerException {@inheritDoc} */ 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; //先要获取入队锁 putLock.lockInterruptibly(); try { /* * 循环检测队列是否已满，如果已满则阻塞 * 考虑这种场景：A入队线程在等待入队条件，B线程消费了1个元素 * 此时刚好C线程获取入队锁，进行入队操作，那么此时A线程被唤醒后 * 必须要再次判断队列是否已满，否则会出现队列元素比容量大的问题 * 所以此处要循环判断 */ while (count.get() == capacity) { notFull.await(); } //入队操作，在锁内部是安全的，修改链表指针位置即可 enqueue(node); //获取入队时的元素个数，并将元素个数加1 c = count.getAndIncrement(); //如果入队后，队列仍未满，则唤醒其他等待写入的线程 //考虑这种场景：A，B两个入队线程都在等待入队条件，C，D两个线程同时消费了2个元素 //此时A，B都可以入队，但是入队操作只会唤醒一次（c==capacity），假如唤醒了A， //那么，A入队后要判断是否还有空间，继续唤醒B if (c + 1 < capacity) notFull.signal(); } finally { putLock.unlock(); } //如果写入时，队列里没有元素，则唤醒出队线程；因为此时可能有出队线程在等待 if (c == 0) signalNotEmpty(); } /** * 向队列尾部插入一个元素，若空间已满则最多等待指定 时间. * * @return {@code true} if successful, or {@code false} if * the specified waiting time elapses before space is available * @throws InterruptedException {@inheritDoc} * @throws NullPointerException {@inheritDoc} */ public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException { if (e == null) throw new NullPointerException(); long nanos = unit.toNanos(timeout); int c = -1; final ReentrantLock putLock = this.putLock; final AtomicInteger count = this.count; //与put一样，都是先获取锁 putLock.lockInterruptibly(); try { while (count.get() == capacity) { //等待超时，返回 if (nanos <= 0) return false; nanos = notFull.awaitNanos(nanos); } enqueue(new Node<E>(e)); c = count.getAndIncrement(); if (c + 1 < capacity) notFull.signal(); } finally { putLock.unlock(); } if (c == 0) signalNotEmpty(); return true; } /** * 向队列尾部插入元素，如果队列已满直接返回fasle；否则，尝试写入，写入成功返回true，失败返回false * @throws NullPointerException if the specified element is null */ 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) notFull.signal(); } } finally { putLock.unlock(); } if (c == 0) signalNotEmpty(); //判断释放写入，如果写入成功c != -1 return c >= 0; }
4.出队操作:
//从队列头部出队，如果队列为空则阻塞 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.signal(); } finally { takeLock.unlock(); } //如果出队前空间已满，则说明此时出队后，还有1个位置，此时唤醒入队线程 if (c == capacity) signalNotFull(); return x; } //从队列头部出队，如果队列为空则阻塞指定的时间，获取成功返回元素，否则返回null public E poll(long timeout, TimeUnit unit) throws InterruptedException { E x = null; int c = -1; long nanos = unit.toNanos(timeout); final AtomicInteger count = this.count; final ReentrantLock takeLock = this.takeLock; //获取出队锁 takeLock.lockInterruptibly(); try { //此处的循环判断原因，同put内部的循环判断 while (count.get() == 0) { //获取超时返回空 if (nanos <= 0) return null; nanos = notEmpty.awaitNanos(nanos); } x = dequeue(); c = count.getAndDecrement(); if (c > 1) notEmpty.signal(); } finally { takeLock.unlock(); } //同上 if (c == capacity) signalNotFull(); return x; } //出队，如果队列为空则返回null 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; } //获取队头元素，此操作不出队，仅仅是读取 public E peek() { //队列为空，返回null if (count.get() == 0) return null; final ReentrantLock takeLock = this.takeLock; //获取出队锁，但是不会出队，仅仅读取 takeLock.lock(); try { //读取队头元素 Node<E> first = head.next; if (first == null) return null; else return first.item; } finally { takeLock.unlock(); } }
5.可以看到JDK内部是通过一个单向链表的数据结构，配合入队的ReentrantLock以及出队的ReentrantLock最终实现了阻塞队列的语义。
