1. 概况

FIFO
初始化时候指定了链表长度，默认为 Integer.MAX_VALUE，而节点是插入时才创建

2. 类定义

继承父类与实现接口与 ArrayBlockingQueue 一样。

public class LinkedBlockingQueue<E> extends AbstractQueue<E>
        implements BlockingQueue<E>, java.io.Serializable {
}

3. 成员变量

与 ArrayBlockingQueue 不一样的地方：

使用两个 Lock 控制入队和出队；ArrayBlockingQueue 只有一个 LOCK 控制并发访问
使用 AtomicInteger 记录当前链表的元素个数；ArrayBlockingQueue 则是使用一个基本类型的变量。

/** 容量大小，默认为 Integer.MAX_VALUE */
private final int capacity;
/** 链表中元素个数 */
private final AtomicInteger count = new AtomicInteger();
/**
 * 链表的头节点，这个头节点是空节点，即 head.item = null
 */
transient Node<E> head;
/**
 * 链表的尾节点，也是一个空节点，即 last.item = null
 */
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();

4. 构造方法

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); // 初始化头节点和尾节点
}
/**
* 将集合中的元素放到链表队列中
*/
public LinkedBlockingQueue(Collection<? extends E> c) {
    this(Integer.MAX_VALUE);
    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");
            enqueue(new Node<E>(e)); // 将元素放到链表尾部
            ++n;
        }
        count.set(n);
    } finally {
        putLock.unlock();
    }
}
// 相当于 last.next = node; last = last.next
private void enqueue(Node<E> node) {
    // assert putLock.isHeldByCurrentThread();
    // assert last.next == null;
    last = last.next = node;
}

5. 成员方法

5.1 入队方法

5.1.1 put

使用 put 锁控制其它线程进行相关入队操作
局部变量 c 用于唤醒正在等待的出队方法

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);
           // getAndIncrement 方法的返回是自增的元素，所以 c 被赋值添加元素之前的链表节点个数；并且 count++
        c = count.getAndIncrement();
        // 添加一个元素还没有满，通知等待队列，还可以进行入队操作
        if (c + 1 < capacity)
            notFull.signal();
    } finally {
        putLock.unlock();
    }
    // 若入队前队列size=0，此时队列元素个数为1，那么就唤醒正在等待出队的方法
    if (c == 0)
        signalNotEmpty(); 
}

5.1.2 offer

与 put 方法类似，都是对 putLock 进行加锁控制
带超时时间的 offer 方法增加了阻塞等待的功能

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();
    return c >= 0;
}
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)
                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;
}

5.2 出队方法

5.2.1 poll

public E poll() {
    final AtomicInteger count = this.count;
    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();  // 先返回再减 1
            if (c > 1)
                notEmpty.signal(); // 通知其它正在等待的 出队方法
        }
    } finally {
        takeLock.unlock();
    }
    if (c == capacity)  // 出队之前队列是满的，现在已经出队了，那么通知其它正在等待的 入队方法
        signalNotFull();
    return x;
}
// 仅在同步方法块中调用
private E dequeue() {
    // assert takeLock.isHeldByCurrentThread();
    // assert head.item == null;
    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;
}
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();
        c = count.getAndDecrement();
        if (c > 1)
            notEmpty.signal();
    } finally {
        takeLock.unlock();
    }
    if (c == capacity)
        signalNotFull();
    return x;
}

5.2.2 peek

public E peek() {
    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.2.3 take

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();
    }
    if (c == capacity)
        signalNotFull();  // 通知正在等待 入队的线程，可以放入元素了
    return x;
}

删除元素

从链表中删除指定元素，删除成功返回 true

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();
    }
}
// 将节点 P 从链表中移除，trail 是 p 节点的前一个节点
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; // 将 last 节点置为 p 的前置节点
    if (count.getAndDecrement() == capacity)
        notFull.signal();
}

6. 内部类

7. 思考

LinkedBlockingQueue 为什么使用两个锁，而 ArrayBlockingQueue 只用一个锁？

首先，使用两个锁的吞吐量肯定比一个锁的大，因为入队和出队并不影响；ArrayBlockingQueue 之所以用一个锁，只能推测了，就是数组实现出队，入队比较简单，再去用锁控制显得多余？

入队操作的时候，如果还没满，方法内容调用的是 notFull 条件队列，通知线程还可以添加；而 ArrayBlockingQueue 添加元素之后调用的是 notEmpty 条件队列，这里为何不一样？

因为 ArrayBlockingQueue 只有一个锁，如果添加元素之后再 notFull 通知正在等待的入队线程，消费线程可能很久以后才会拿到锁。

入队方法结尾，c 为什么不用加锁通知？为什么是等于 0 或者等于 captive 而不是大于 0

c == 0 时，表明入队方法前队列为空，此时很有可能有消费线程在等待，所以得使用 signalNotEmpty 通知正在等待的消费线程

c > 0 时，队列是有数据的，此时消费线程可能都在消费中，此时通知的意义并不大。

Java 后端技能专题

LinkedBlockingQueue 源码学习