1. 变量和构造方法
private final int capacity; // 容量（最大节点个数），默认Integer.MAX_VALUE private final AtomicInteger count = new AtomicInteger(); // 当前节点个数 transient Node<E> head; // 队列头结点 private transient Node<E> last; // 队列尾节点 private final ReentrantLock takeLock = new ReentrantLock(); // 消费锁 private final Condition notEmpty = takeLock.newCondition(); // 队列非空 private final ReentrantLock putLock = new ReentrantLock(); // 生产锁 private final Condition notFull = putLock.newCondition(); // 队列非满 public LinkedBlockingQueue() { this(Integer.MAX_VALUE); } public LinkedBlockingQueue(int capacity) { if (capacity <= 0) throw new IllegalArgumentException(); this.capacity = capacity; last = head = new Node<E>(null); // 添加队列初始头节点（last = head表示队列为空） } public LinkedBlockingQueue(Collection<? extends E> c) { this(Integer.MAX_VALUE); final ReentrantLock putLock = this.putLock; putLock.lock(); // 加生产锁 try { int n = 0; for (E e : c) { if (e == null) throw new NullPointerException(); if (n == capacity) // 当前节点个数超过容量 throw new IllegalStateException("Queue full"); enqueue(new Node<E>(e)); ++n; } count.set(n); } finally { putLock.unlock(); // 释放生产锁 } }
2. 生产（put、offer）
static class Node<E> { // 队列节点 E item; Node<E> next; Node(E x) { item = x; } } private void enqueue(Node<E> node) { // 队列非满时，节点入队列（在队列尾部添加节点） last = last.next = node; } private void signalNotEmpty() { // 唤醒等待非空的线程 final ReentrantLock takeLock = this.takeLock; takeLock.lock(); try { notEmpty.signal(); } finally { takeLock.unlock(); } } 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 { while (count.get() == capacity) { // 队列满 notFull.await(); // 等待队列非满 } enqueue(node); // node入队列 c = count.getAndIncrement(); if (c + 1 < capacity) // node入队列后，队列非满 notFull.signal(); // 唤醒等待队列非满的线程 } finally { putLock.unlock(); // 释放生产锁 } if (c == 0) // node入队列前，队列为空 signalNotEmpty(); // 唤醒等待非空的节点 } 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; putLock.lockInterruptibly(); // 加生产锁 try { while (count.get() == capacity) { // 队列满 if (nanos <= 0) // 超时（awaitNanos可能已被signal，但在SyncQueue中排队等锁时超时，见ReentrantLock） return false; nanos = notFull.awaitNanos(nanos); // 等待队列非满（在nanos时间内） } enqueue(new Node<E>(e)); // new Node入队列 c = count.getAndIncrement(); if (c + 1 < capacity) // new Node入队列后，队列非满 notFull.signal(); // 唤醒等待队列非满的线程 } finally { putLock.unlock(); // 释放生产锁 } if (c == 0) // new Node入队列前，队列为空 signalNotEmpty(); // 唤醒等待 return true; }
3. 消费（take、poll）
private E dequeue() { // 队列非空时，节点出队列（移除head.next） Node<E> h = head; Node<E> first = h.next; h.next = h; // help GC head = first; E x = first.item; first.item = null; return x; } private void signalNotFull() { // 唤醒等待非满的线程 final ReentrantLock putLock = this.putLock; putLock.lock(); try { notFull.signal(); } finally { putLock.unlock(); } } 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(); // head.next出队列 c = count.getAndDecrement(); if (c > 1) // head.next出队列后，队列非空 notEmpty.signal(); // 唤醒等待队列非空的线程 } finally { takeLock.unlock(); // 释放消费锁 } if (c == capacity) // head.next出队列前，队列已满 signalNotFull(); // 唤醒等待队列非满的线程 return x; } 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 { while (count.get() == 0) { // 队列空 if (nanos <= 0) // 超时 return null; nanos = notEmpty.awaitNanos(nanos); // 等待队列非空 } x = dequeue(); // head.next出队列 c = count.getAndDecrement(); if (c > 1) // head.next出队列后，队列非空 notEmpty.signal(); // 唤醒等待队列非空的线程 } finally { takeLock.unlock(); // 释放消费锁 } if (c == capacity) // head.next出队列前，队列已满 signalNotFull(); // 唤醒等待队列非满的线程 return x; }
4. 加全锁（remove）
void fullyLock() { // 加全锁（remove、contains、toArray、clear、Itr） putLock.lock(); takeLock.lock(); } void fullyUnlock() { // 释放全锁 takeLock.unlock(); putLock.unlock(); } void unlink(Node<E> p, Node<E> trail) { p.item = null; trail.next = p.next; if (last == p) last = trail; if (count.getAndDecrement() == capacity) // 移除节点前，队列已满 notFull.signal(); // 唤醒等待队列非满的线程 } 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) { // 遍历队列（链表），p为当前节点，trai为p前置节点 if (o.equals(p.item)) { unlink(p, trail); return true; } } return false; } finally { fullyUnlock(); // 释放全锁 } }