基本的入队出队
public class LinkedBlockingQueue<E> extends AbstractQueue<E>
implements BlockingQueue<E>, java.io.Serializable {
static class Node<E> {
E item;
/**
* 下列三种情况之一
* - 真正的后继节点
* - 自己, 发生在出队时
* - null, 表示是没有后继节点, 是最后了
*/
Node<E> next;
Node(E x) { item = x; }
}
}
初始化链表 last = head = new Node<E>(null); Dummy 节点用来占位，item 为 null
当一个节点入队 last = last.next = node;
再来一个节点入队 last = last.next = node;
出队：【移除头节点dummy（只是占位的），next指向自己，为了帮助垃圾回收】
Node<E> h = head; Node<E> first = h.next; h.next = h; // help GC head = first; //因为dummy只是占位的 E x = first.item; first.item = null;//头节点重新变成dummy return x;
加锁分析
高明之处在于用了两把锁和 dummy 节点
用一把锁，同一时刻，最多只允许有一个线程（生产者或消费者，二选一）执行
用两把锁（锁住队列头和尾），同一时刻，可以允许两个线程同时（一个生产者与一个消费者）执行
消费者与消费者线程仍然串行【生产者和消费者不相关】
生产者与生产者线程仍然串行
线程安全分析
当节点总数大于 2 时（包括 dummy 节点），putLock 保证的是 last 节点的线程安全，takeLock 保证的是head 节点的线程安全。两把锁保证了入队和出队没有竞争
当节点总数等于 2 时（即一个 dummy 节点（防止锁住同一个对象），一个正常节点【占位】）这时候，仍然是两把锁锁两个对象，不会竞争
当节点总数等于 1 时（就一个 dummy 节点）这时 take 线程会被 notEmpty 条件阻塞，有竞争，会阻塞
// 用于 put(阻塞) offer(非阻塞)队尾 private final ReentrantLock putLock = new ReentrantLock(); // 用户 take(阻塞) poll(非阻塞)队头 private final ReentrantLock takeLock = new ReentrantLock();
put：
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;
// count 用来维护元素计数
final AtomicInteger count = this.count;
///上锁，可被打断
putLock.lockInterruptibly();
try {
// 满了等待
while (count.get() == capacity) {
// 倒过来读就好: 等待 notFull
notFull.await();
}
// 有空位, 入队（两次赋值操作）且计数加一
enqueue(node);
///c返回的是+1前的数？
c = count.getAndIncrement();
// 除了自己 put 以外, 队列还有空位, 由自己叫醒其他 put 线程
if (c + 1 < capacity)
notFull.signal();
} finally {
putLock.unlock();
}
// 如果队列中有一个元素, 叫醒 take 线程。非必要才用，因为生产者可以自己唤醒自己的线程
if (c == 0)
// 这里调用的是 notEmpty.signal() 而不是 notEmpty.signalAll() 是为了减少竞争
signalNotEmpty();
}
take:注意，这种情况是take之前，队列已经满了，之前的所有生产者都等待呢。如果c==capacity不成立，那么生产者没有阻塞，可以自己唤醒其他生产者
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();
}
// 如果队列中只有一个空位时, 叫醒 put 线程
// 如果有多个线程进行出队, 第一个线程满足 c == capacity, 但后续线程 c < capacity
/注意，这种情况是take之前，队列已经满了，之前的所有生产者都等待呢。如果c==capacity不成立，那么生产者没有阻塞，可以自己唤醒其他生产者
if (c == capacity)
// 这里调用的是 notFull.signal() 而不是 notFull.signalAll() 是为了减少竞争
signalNotFull();
return x;
}
由 put 唤醒 put 是为了避免信号不足
性能比较
主要列举 LinkedBlockingQueue 与 ArrayBlockingQueue 的性能比较
Linked 支持有界，Array 强制有界
Linked 实现是链表，Array 实现是数组
Linked 是懒惰的，而 Array 需要提前初始化 Node 数组
Linked 每次入队会生成新 Node，而 Array 的 Node 是提前创建好的
Linked 两把锁，Array 一把锁