从源码全面解析 LinkedBlockingQueue 的来龙去脉

add(E)     	// 添加数据到队列，如果队列满了，无法存储，抛出异常offer(E)    // 添加数据到队列，如果队列满了，返回falseoffer(E,timeout,unit)   // 添加数据到队列，如果队列满了，阻塞timeout时间，如果阻塞一段时间，依然没添加进入，返回falseput(E)      // 添加数据到队列，如果队列满了，挂起线程，等到队列中有位置，再扔数据进去，死等！

remove()    // 从队列中移除数据，如果队列为空，抛出异常poll()      // 从队列中移除数据，如果队列为空，返回null，么的数据poll(timeout,unit)   // 从队列中移除数据，如果队列为空，挂起线程timeout时间，等生产者扔数据，再获取take()     // 从队列中移除数据，如果队列为空，线程挂起，一直等到生产者扔数据，再获取

public class LinkedBlockingQueueTest {    public static void main(String[] args) throws Exception {        LinkedBlockingQueue queue = new LinkedBlockingQueue();
        // 生产者扔数据        queue.add("1");        queue.offer("2");        queue.offer("3", 2, TimeUnit.SECONDS);        queue.put("2");
        // 消费者取数据        System.out.println(queue.remove());        System.out.println(queue.poll());        System.out.println(queue.poll(2, TimeUnit.SECONDS));        System.out.println(queue.take());    }}

LinkedBlockingQueue queue = new LinkedBlockingQueue();
// 如果我们不指定容量大小的话,这里的容量默认为Integer.MAX_VALUEpublic LinkedBlockingQueue() {    this(Integer.MAX_VALUE);}
public LinkedBlockingQueue(int capacity) {    // 如果容量传进来是小于等于0的,直接抛异常    if (capacity <= 0){        throw new IllegalArgumentException();    }    // 当前的容量赋值    this.capacity = capacity;    // 这里其实和我们的AQS有点像    // 搞一个虚拟的头结点,减少后面的判空    last = head = new Node<E>(null);}

class Node<E> {    // 当前的数据    E item;    // 指向下一个数据的指针    Node<E> next;    Node(E x) {        item = x;    }}
// 当前链表中存在的数据数量private final AtomicInteger count = new AtomicInteger();
// 读锁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 boolean add(E e) {    return super.add(e);}
// 走到这里会发现，我们的add方法就是调用了offer方法// offer: 添加数据到队列，如果队列满了，返回false// 所以这里offer满了，就会抛出异常："Queue full"public boolean add(E e) {    if (offer(e))        return true;    else        throw new IllegalStateException("Queue full");}

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 {        // 如果当前数据量小于总容量        // 这里我们上面也检查过，相当于DCL的意思        if (count.get() < capacity) {            // 插入队列            enqueue(node);            // 得到当前数据量            // 这里需要注意：getAndIncrement先返回数据，再加一            c = count.getAndIncrement();            // 如果我们发现当前数据量还小于总容量            // 也就是我们可以继续放数据            if (c + 1 < capacity)                // 唤醒其他的生产者线程扔数据                // 当然这里稍微多说一点，这里的唤醒指的是将生产者从Condition队列放到AQS队列中                // 具体什么时候执行还需要看AQS的调度                notFull.signal();        }    } finally {        // 解锁        putLock.unlock();    }    // 如果我们当前数据量为0,代表队列中原来无数据    // 但上面现在扔进去了一个    if (c == 0)        // 需要唤醒所有的消费者消费数据        signalNotEmpty();    return c >= 0;}
private void enqueue(Node<E> node) {    // 将当面节点挂在last节点后    // 将last节点指向当前节点    last = last.next = node;}

// 这里我们的Condition聊过// 必须持有当前锁资源才可以使用Condition的方法private void signalNotEmpty() {    // 拿到读锁    final ReentrantLock takeLock = this.takeLock;    // 加锁    takeLock.lock();    try {        // 唤醒消费者线程        notEmpty.signal();    } finally {        // 解锁        takeLock.unlock();    }}

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 {        // 如果当前数据量小于总容量        // 这里我们上面也检查过，相当于DCL的意思        while (count.get() == capacity) {            // 如果我们剩余时间小于0,直接失败即可            if (nanos <= 0)                return false;            // 反之生产者线程写入挂起nanos时间            nanos = notFull.awaitNanos(nanos);        }        // 添加至队列        enqueue(new Node<E>(e));        // 得到当前数据量        // 这里需要注意：getAndIncrement先返回数据，再加一        c = count.getAndIncrement();        // 如果我们发现当前数据量还小于总容量        // 也就是我们可以继续放数据        if (c + 1 < capacity)            // 唤醒其他的生产者线程扔数据            // 当然这里稍微多说一点，这里的唤醒指的是将生产者从Condition队列放到AQS队列中            // 具体什么时候执行还需要看AQS的调度            notFull.signal();    } finally {        // 解锁        putLock.unlock();    }    // 如果我们当前数据量为0,代表队列中原来无数据    // 但现在扔进去了一个,唤醒消费者线程    if (c == 0)        signalNotEmpty();    return true;}

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);        c = count.getAndIncrement();        if (c + 1 < capacity)            notFull.signal();    } finally {        putLock.unlock();    }    if (c == 0)        signalNotEmpty();}

public E remove() {    E x = poll();    if (x != null)        return x;    else        throw new NoSuchElementException();}

public E poll() {    // 获取当前链表的数据量    final AtomicInteger count = this.count;    // 如果数据量为0,说明无数据    // 消费者无法消费，直接返回null即可    if (count.get() == 0)        return null;    E x = null;    int c = -1;    // 拿到读锁    final ReentrantLock takeLock = this.takeLock;    // 加锁    takeLock.lock();    try {        // 如果数据量大于0,说明有数据        // 这里我们上面也检查过，相当于DCL的意思        if (count.get() > 0) {            // 取数            x = dequeue();            // 得到当前数据量            // 这里需要注意：getAndIncrement先返回数据，再减一            c = count.getAndDecrement();            // 如果我们的数据量大于1,则唤醒消费者来消费            if (c > 1)                notEmpty.signal();        }    } finally {        // 解锁        takeLock.unlock();    }    // 如果数据量等于当前的总容量    // 说明当前的链表已经有空余了,唤醒生产者生产    if (c == capacity)        signalNotFull();    return x;}
// 这个取数据和我们的AQS有点像// 去除当前数据并且将当前节点作为头结点private E dequeue() {    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 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 {        // 如果当前的数据量为0        while (count.get() == 0) {            // 如果时间没有剩余,直接返回null即可            if (nanos <= 0)                return null;            // 让消费者线程等待nanos时间            nanos = notEmpty.awaitNanos(nanos);        }        // 取数据        x = dequeue();        // 后面都是一样的        c = count.getAndDecrement();        if (c > 1)            notEmpty.signal();    } finally {        takeLock.unlock();    }    if (c == capacity)        signalNotFull();    return x;}

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;}