【JUC源码】阻塞队列：LinkedBlockingQueue 源码分析

分手后的思念是犯贱 2022-12-11 05:23 244阅读 0赞

> 阻塞队列系列：
> 
>  *  [【JUC源码】阻塞队列：继承关系及接口分析][JUC]
>  *  [【JUC源码】阻塞队列：ArrayBlockingQueue 源码分析][JUC_ArrayBlockingQueue]
>  *  [【JUC源码】阻塞队列：LinkedBlockingQueue 源码分析][JUC_LinkedBlockingQueue]
>  *  [【JUC源码】阻塞队列：DelayQueue 源码分析][JUC_DelayQueue]
>  *  [【JUC源码】阻塞队列：SynchronousQueue 源码分析（非公平模式）][JUC_SynchronousQueue]
>  *  [【JUC源码】阻塞队列：SynchronousQueue 源码分析（公平模式）][JUC_SynchronousQueue 1]
>  *  [【JUC源码】阻塞队列：总结 --特性对比&应用场景举例][JUC_ --]
>  *  [【JUC源码】阻塞队列：各种阻塞队列面试题合集][JUC 1]

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;
    
        // 链表已有元素多少，使用 AtomicInteger，所以是线程安全的
        private final AtomicInteger count = new AtomicInteger();
    
        // 链表头
        // 注；head一般指向的都是一个哨兵节点（数据为null）。
        //    1.目的：辅助出队，即在 take/poll/remove 需要将头结点删除（dequeue）时发挥作用。
        //    2.注意：这个哨兵节点是随着每次出队都变化的。具体请看后面dequeue方法注释
        transient Node<E> head;
    
        // 链表尾
        private transient Node<E> last;
    	
    	//--------------------------------锁---------------------------------------
        // take 时的锁
        private final ReentrantLock takeLock = new ReentrantLock();
        
        // take 的条件队列，其实很容易理解：出队时队列不能为空
        private final Condition notEmpty = takeLock.newCondition();
    
        // put 时的锁
        // 注：设计两把锁的目的，主要为了 take 和 put 可以同时进行。ArrayBlockingQueue只有一把锁。
        private final ReentrantLock putLock = new ReentrantLock();
        
        // put 的条件队列，其实很容易理解：入队时队列不能是满的
        private final Condition notFull = putLock.newCondition();
     
        // 迭代器。LinkedBlockingQueue 实现了自己的迭代器
        private class Itr implements Iterator<E> {
        }
        
        //--------------------------------构造器--------------------------------------
        // 构造函数一：空参构造。不指定容量时，默认 Integer 的最大值
        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();
                    // capacity 代表链表的大小，在这里是 Integer 的最大值
                    // 如果集合类的大小大于 Integer 的最大值，就会报错
                    // 其实这个判断完全可以放在 for 循环外面，这样可以减少 Integer 的最大值次循环(最坏情况)
                    if (n == capacity)
                        throw new IllegalStateException("Queue full");
                    enqueue(new Node<E>(e));
                    ++n;
                }
                count.set(n);
            } finally {
        
                putLock.unlock();
            }
        }
    }

*  链表的作用是为了保存当前节点，节点中的数据可以是任意东西，是一个泛型，比如说队列被应用到线程池时，节点就是线程（thread），比如队列被应用到消息队列中，节点就是消息，节点的含义主要看队列被使用的场景
 *  锁有 take 锁和 put 锁，是为了保证队列操作时的线程安全，设计两种锁，是为了 take 和 put 两种操作可以同时进行，互不影响。如果只有一把锁 put，take 同时只能进行一个
 *  初始化时，容量大小是不会影响性能的，只影响在后面的使用，因为初始化队列太小，容易导致没有放多少就会报队列已满的错误

另外，在对给定集合数据进行初始化时，我们不反对在每次 for 循环的时候，都去检查当前链表的大小是否超过容量，但我们希望在 for 循环开始之前就做一步这样的工作。举个列子，给定集合大小是 1 w，链表大小是 9k，按照现在代码实现，只能在 for 循环 9k 次时才能发现，原来给定集合的大小已经大于链表大小了，导致 9k 次循环都是在浪费资源，还不如在 for 循环之前就 check 一次，如果 1w > 9k，直接报错即可

> PS：队列的主要方法无非就三个：入队、出队、队首。所以接下来我们就从这三个方法入手，来看看 LinkedBlockingQueue 的具体实现。这里的内容可以对比着 ArrayBlockingQueue 的源码进行理解，具体参考[【JUC源码】ArrayBlockingQueue源码分析][JUC_ArrayBlockingQueue]，两者的实现思路大同小异。

## 1.入队 ##

### put()：满时阻塞 ###

大致流程如下；

1.  首先是一些准备操作，比如新建 node，获取到 put 锁和计数器 count
2.  对 putLock 加锁，所以后面的新增数据是线程安全的
3.  新增数据分为两步：
    
     *  如果队列满了，当前线程会被加入条件队列中阻塞
     *  将新node简单的追加到链表的尾部
4.  新增数据成功后，在适当时机，会唤醒阻塞的 put 和 take 线程，保证了唤起的时机不被浪费
    
     *  队列不满，唤起 put 的等待线程
     *  在 put 之前队列为空会，唤醒 take 的等待线程

// 把e新增到队列的尾部。如果有可以新增的空间的话，直接新增成功，否则当前线程陷入等待
    public void put(E e) throws InterruptedException {
        
        // 要添加的元素 e 为空，抛出异常
        if (e == null) throw new NullPointerException();
        // 预先设置 c 为 -1，约定负数为新增失败
        int c = -1;
        
        Node<E> node = new Node<E>(e); // 创建新node
    	
        final ReentrantLock putLock = this.putLock; // 获取put的锁
        final AtomicInteger count = this.count; // 获取到队列中元素的个数
         
        putLock.lockInterruptibly(); // 加锁，设置可中断锁
        try {
        
            // ！！！队列如果是满的，就将当前线程加入到notFull的条件队列中，然后进入阻塞状态
            // 等待某个线程[新增成功 || 拿走队列元素 || 删除元素]后，有机会被唤醒进入同步队列
            // 注：这里的while循环有double-check的意思，即防止线程已经被调度执行了，但前一刻有另一个线程put/offer/add成功
            while (count.get() == capacity) {
        
                // await 让出CPU，休眠，释放锁
                notFull.await();
            }
    
            // 队列没有满，直接新增到队列的尾部
            enqueue(node);
    
            // conut+1。这里是原子操作新增，getAndIncrement 返回的是旧值，所以 c 是比真实的 count 小 1 的
            c = count.getAndIncrement();
    
            // 如果链表现在的大小 小于链表的容量，说明队列未满。可以尝试唤醒一个 put 的等待线程       
            if (c + 1 < capacity)
                notFull.signal();
    
        } finally {
        
            putLock.unlock(); // 释放锁
        }
        // c==0，代表了队列之前空，现在刚刚新增了一个的情况。所以会尝试唤醒一个take的等待线程
        if (c == 0)
            signalNotEmpty();
    }

> 注：从上面代码可以看出，不像 ArrayBlockingQueue 的 put 线程唤醒 take 线程，LinkedBlockingQueue 是新 put 线程唤醒旧 put 线程，即每个 put 线程执行到最后都会判断一下队列是否满了，如果没满就尝试唤醒一个 put 线程。  
> 所以，虽然 putLock 和 takeLock 两个锁提高了入队/出队的并发能力，但是也牺牲了一定的响应速度（因为在wait的线程需要后来的新线程notify）

### offer()：满时返回false ###

public boolean offer(E e) {
        
            if (e == null) throw new NullPointerException();
            final AtomicInteger count = this.count;
        	// ！！！先判断是否满，满则返回false
            if (count.get() == capacity)
                return false;
    		// 剩下逻辑同 put          
            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;
        }

### add()：队满抛异常 ###

在 LinkedBlockingQueue 中并没有直接实现 add 方法，它是在父类 AbstractQueue 中，逻辑很简单就是简单调用 offer方法  
![在这里插入图片描述][20200928184846539.png_pic_center]

### enqueue() ###

*  入队，把新元素放到队尾
 *  add，offer，put都要调用此方法

private void enqueue(Node<E> node) {
        
    	// 1.连接：last.next = node
    	// 2.后移：last = node
        last = last.next = node;
    }

## 2.出队 ##

### take()：队空阻塞 ###

队列空则阻塞，方法的原理与 put 相似

1.  先上锁，所以后面的出队操作线程安全
2.  取数据时分为两步：
    
    1.  若队列为空，当前线程会被加入 notEmpty 条件队列阻塞
    2.  调用dequeue，头删
3.  出队完成后，看能否唤醒阻塞的线程
    
     *  队列不为空，唤醒 take 的等待线程
     *  在 take 之前满队列满，唤醒 put 的等待线程

public E take() throws InterruptedException {
        
        E x;
        // 默认负数，代表失败
        int c = -1;
        
        final AtomicInteger count = this.count; // count 代表当前链表数据的真实大小
        final ReentrantLock takeLock = this.takeLock; // 获取take的锁
        
        takeLock.lockInterruptibly(); // 加锁，设置为可中断
        try {
        
            // ！！！队列如果是空的，就将当前线程加入到notEmpty条件队列中，然后进入阻塞状态
            // 等待某个线程拿向队列放入元素后，有机会被唤醒进入同步队列
            // 注：这里的while循环有double-check的意思，即防止线程已经被调度执行了，但前一刻有另一个线程take/poll/remove成功了，所以他又要进入条件队列notFull中阻塞等待。
            while (count.get() == 0) {
        
                notEmpty.await();
            }
           
            // 非空队列，从队列的头部拿一个出来
            x = dequeue();
            
            // count-1，这里是原子操作，getAndDecrement 返回的值是旧值（c 比真实的 count 大1）
            c = count.getAndDecrement();
            
            // 如果队列里面有值，从 take 的等待线程里面唤醒一个。
            if (c > 1)
                notEmpty.signal();
        } finally {
        
            takeLock.unlock(); // 释放锁
        }
        // 如果队列空闲还剩下一个，尝试从 put 的等待线程中唤醒一个
        if (c == capacity)
            signalNotFull();
        return x;
    }

### poll()：队空返回null ###

public E poll() {
        
            final AtomicInteger count = this.count;
            // !!!队列为空返回null
            if (count.get() == 0)
                return null;
            E x = null;
            // 下面逻辑同 put
            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;
        }

### remove()：队空抛异常 ###

同上面的 add，remove 方法在 LinkedBlockingQueue 也没有具体实现，也是在父类 AbstractQueue 中，也是直节调用 poll  
![在这里插入图片描述][20200928190657551.png_pic_center]

### dequeue() ###

*  队头中取数据，链表头删。这里就体现出了哨兵节点的作用，具体请看下面注释。
 *  remove，poll，take 都会调用此方法

private E dequeue() {
        
        Node<E> h = head; // 获取头结点h（哨兵节点）
        Node<E> first = h.next; // 获取第一个元素节点first
        h.next = h; // help GC
        head = first; // 将头结点置为first。到这步相当于已经删除了之前的哨兵节点
        E x = first.item; // 保存第一元素节点first的数据x
        first.item = null;// 将first的数据删除，变为新的哨兵节点
        return x;
    }

## 3.获取队首：peek ##

*  查看并不删除元素，如果队列为空，返回 null
 *  注意，虽然删除，但在读取的时候也要拿到 take 锁，避免头结点被删除

public E peek() {
        
        // count 代表队列实际大小，队列为空，直接返回 null
        if (count.get() == 0)
            return null;
        final ReentrantLock takeLock = this.takeLock;
       	// 上take锁，避免读取时被删除
        takeLock.lock();
        try {
        
            // 拿到队列头
            Node<E> first = head.next;
            // 判断队列头是否为空，并返回
            if (first == null)
                return null;
            else
                return first.item;
        } finally {
        
            takeLock.unlock(); // 释放锁
        }
    }

[JUC]: https://blog.csdn.net/weixin_43935927/article/details/113952868
[JUC_ArrayBlockingQueue]: https://blog.csdn.net/weixin_43935927/article/details/108833582
[JUC_LinkedBlockingQueue]: https://blog.csdn.net/weixin_43935927/article/details/108853656
[JUC_DelayQueue]: https://blog.csdn.net/weixin_43935927/article/details/108858129
[JUC_SynchronousQueue]: https://blog.csdn.net/weixin_43935927/article/details/108866902
[JUC_SynchronousQueue 1]: https://blog.csdn.net/weixin_43935927/article/details/113964294
[JUC_ --]: https://blog.csdn.net/weixin_43935927/article/details/108888294
[JUC 1]: https://blog.csdn.net/weixin_43935927/article/details/108857735
[20200928184846539.png_pic_center]: /images/20221123/c8e602073d344cf0b5271fc555ba8993.png
[20200928190657551.png_pic_center]: /images/20221123/1aa3834e6954427da76c0497ddec07da.png