【集合】LinkedBlockingQueue 源码分析

痛定思痛。 2023-02-15 03:41 125阅读 0赞

# 前言 #

Github：[https://github.com/yihonglei/jdk-source-code-reading][https_github.com_yihonglei_jdk-source-code-reading]（java-concurrent）

# 一 概述 #

LinkedBlockingQueue 链式队列，底层基于单向链表实现的队列，可以想象成生活中排队一个道理，BlockingQueue 类关系图。

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lobF9qeHk_size_16_color_FFFFFF_t_70][]

BlockingQueue 上层跟普通 List 一样，遵循集合的标准规范，这篇文章主要分析 LinkedBlockingQueue。

**LinkedBlockingQueue**：一个由链表结构实现的无界阻塞队列，因为默认容量大小为 Integer.MAX\_VALUE。

# 二 LinkedBlockingQueue 实例 #

这里分析一下 LinkedBlockingQueue 的存入元素和取出元素源码，因为线程池用得比较多。先上个简单实例，感受下什么是阻塞。

package com.jpeony.concurrent.queue;
    
    import java.util.concurrent.LinkedBlockingQueue;
    
    /**
     * @author yihonglei
     */
    public class LinkedBlockingQueueTest {
        public static void main(String[] args) throws InterruptedException {
            // 创建队列
            LinkedBlockingQueue blockingQueue = new LinkedBlockingQueue(10);
            // 存入元素
            blockingQueue.put("object01");
            // 循环取元素
            while (true) {
                System.out.println("开始取值...");
                // 当队列没有元素的时候，队里阻塞当前线程，直到队列不为空
                System.out.println(blockingQueue.take());
                System.out.println("取值完成！");
            }
        }
    }

**运行结果**

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lobF9qeHk_size_16_color_FFFFFF_t_70 1][]

**程序说明**

1）创建一个阻塞队列，初始化大小为  10；

2）存入一个元素；

3）通过死循环去队列取值，第一轮因为队列有元素，顺利取到值，第二轮，因为队列没有元素，

所以 main 线程被一直阻塞了，因为这里没有别的地方放入元素去唤醒阻塞线程，所以就一直阻塞。

# 三 LinkedBlockingQueue 源码分析 #

主要分析 LinkedBlockingQueue 的 offer() 非阻塞入队 和 put() 阻塞入队、take()  阻塞出队 和 poll() 非阻塞出队。

# 重要属性 #

/**
     * 链表节点
     */
    static class Node<E> {
    	// 元素值
        E item;
        // 下一个节点引用
        Node<E> next;
    
        Node(E x) { item = x; }
    }
    
    // 最大容量
    private final int capacity;
    
    // 队列元素个数计数器
    private final AtomicInteger count = new AtomicInteger();
    
    // 链表头结点
    transient Node<E> head;
    
    // 链表尾结点
    private transient Node<E> last;
    
    // take 或 poll 时加锁
    private final ReentrantLock takeLock = new ReentrantLock();
    
    // take 时队列为空，线程被阻塞，当队列不为空时，唤醒被阻塞的线程继续获取元素
    private final Condition notEmpty = takeLock.newCondition();
    
    // put 或 offer 时加锁
    private final ReentrantLock putLock = new ReentrantLock();
    
    // put 时队列满了，线程被阻塞，当不为空时，唤醒被阻塞线程继续尝试入队
    private final Condition notFull = putLock.newCondition();

## LinkedBlockingQueue 构造器 ##

public LinkedBlockingQueue() {
    	// 链表容量默认为 Interger 的最大值
        this(Integer.MAX_VALUE);
    }
    
    public LinkedBlockingQueue(int capacity) {
    	// 容量必须大于等于 1，否则抛出不合法异常
        if (capacity <= 0) throw new IllegalArgumentException();
        // 链式队列容量
        this.capacity = capacity;
        // 初始化链表头结点和尾节点
        last = head = new Node<E>(null);
    }

1）构造器如果不传入 capacity 大小，capacity 默认是 int 的最大值；

2）初始化了链表的头结点和尾结点；

## LinkedBlockingQueue\#put()/offer() ##

### **offer() ** ###

非阻塞，队列满了就放不进去，也不会阻塞等着有空位置的时候再入队，直接返回放入失败；

JDK 默认线程池用的 offer()，线程池关于核心线程创建、队列、最大线程、饱和策略，可以参考：[线程池实现原理][Link 1]；

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);
                // 队列元素加1
                c = count.getAndIncrement();
                // 如果队列不满，唤醒哪些添加被阻塞的线程尝试添加元素，主要针对put()操作的，put操作添加队列满时，线程会被挂起阻塞，一直等待入队
                if (c + 1 < capacity)
                    notFull.signal();
            }
        } finally {
        	// 释放锁
            putLock.unlock();
        }
        // 添加元素成功，唤醒take()被阻塞的线程从队列获取元素消费，只需要唤醒一次，不用每次添加成功都去唤醒，因为只要队列有值，
        // take() 的线程就不会被阻塞；
        // 这里用0判断，感到是不是很奇怪，一开始 c = -1，当第一次添加元素的时候 c = count.getAndIncrement(); 返回的是0
        // 即 count.getAndIncrement() 返回的是自增前的值，从0开始，用0判断就是为了只唤醒一次，如果队列为空了，线程被阻塞了，
        // 也是重复添加第一次的流程唤醒take()被阻塞的线程。
        if (c == 0)
            signalNotEmpty();
        return c >= 0;
    }

ReentrantLock 加锁机制，以及如何实现重入锁的，如何实现条件等待唤醒的，可以参考：[ReentrantLock实现原理][ReentrantLock]。

**enqueue() 入队**

private void enqueue(Node<E> node) {
        // assert putLock.isHeldByCurrentThread();
        // assert last.next == null;
        // 入队分两步操作：
       	// 1. last.next = node，即 将新节点添加到队尾
       	// 2. last = last.next，即 队尾元素变为新节点，即尾部指针向后移动
        last = last.next = node;
    }

### **put()** ###

阻塞，当队列满了，一直等着啥时候空了，啥时候放，很执着；

public void put(E e) throws InterruptedException {
        if (e == null) throw new NullPointerException();
        // Note: convention in all put/take/etc is to preset local var
        // holding count negative to indicate failure unless set.
        int c = -1;
        Node<E> node = new Node<E>(e);
        final ReentrantLock putLock = this.putLock;
        final AtomicInteger count = this.count;
        putLock.lockInterruptibly();
        try {
            /*
             * Note that count is used in wait guard even though it is
             * not protected by lock. This works because count can
             * only decrease at this point (all other puts are shut
             * out by lock), and we (or some other waiting put) are
             * signalled if it ever changes from capacity. Similarly
             * for all other uses of count in other wait guards.
             */
            // 如果队列已满，则添加队列的线程被阻塞，直到队列有空闲被唤醒继续执行尝试重新添加队列，如此反复，直到添加成功为止
            while (count.get() == capacity) {
                notFull.await();
            }
            // 入队
            enqueue(node);
            c = count.getAndIncrement();
            // 唤醒
            if (c + 1 < capacity)
                notFull.signal();
        } finally {
            putLock.unlock();
        }
        if (c == 0)
            signalNotEmpty();
    }

通过源码分析，offer() 和 put() 最大的区别在于 offer()  入队时，如果队列已满，直接就退出添加，return 了，

不去阻塞等待，干净利落，而 put() 看着队列已满了，就阻塞等待，等待啥时候有空位置了，才尝试入队，

如此反复，直到入队为止。

offer() 就像生活中排队买票一样，发现排得满满的，直接不排了，想别的办法解决，而 put() 则相反，发现排满了，

保安说老弟不慌，你先抽支烟等待，啥时候空了，我叫你，添加到队尾，但是不止你一个人抽烟，所以我们需要加锁机制，

否则，空出来一个位置，一批人去抢，容易打起来，所以对于这种情况，又分为公平锁和非公平锁，非公平锁的，

看谁跑得快，是公平锁的，就讲个先来后到。

## LinkedBlockingQueue\#take()/poll() ##

### **take()** ###

阻塞，队列为空时，线程阻塞，一直等到队列有值获取，也是很执着；

public E take() throws InterruptedException {
        E x;
        int c = -1;
        final AtomicInteger count = this.count;
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lockInterruptibly();
        try {
        	// 如果队列为空，则线程阻塞，直到队列不为空，与offer() 或 put() 唤醒上下文相呼应，逻辑上就是一个“生产者-消费者”模型。
            while (count.get() == 0) {
                notEmpty.await();
            }
            // 出队
            x = dequeue();
            c = count.getAndDecrement();
            if (c > 1)
            	// 队列还有元素，继续唤醒阻塞线程获取元素
                notEmpty.signal();
        } finally {
            takeLock.unlock();
        }
        // 因为上面取出了元素，留出空的位置，唤醒添加队列被阻塞的线程，可以继续添加了
        // 这里是不是有疑问，为什么是等于，不是小于呢？因为如果 队列有三个元素，c = count.getAndDecrement(); 返回的是3，即返回取出之前的容量，
        // 取出之后实际元素个数就变为2了，因为如果有空，也只需要唤醒一次，这个代码也可以改成 c -1 < capacity，但是会导致唤醒多次，是浪费，
        // 所以为了，控制一次唤醒，用等于即可
        if (c == capacity)
            signalNotFull();
        return x;
    }

**dequeue() 出队**

private E dequeue() {
        // assert takeLock.isHeldByCurrentThread();
        // assert head.item == null;
        // 头结点（带头链表，头结点并没有存元素，只是一个哨兵结点，跟LinkedList不一样，LinkedList 是不带头链表）
        Node<E> h = head;
        // 头结点的下一个节点，即将作为新的头节点，因为头节点要出队
        Node<E> first = h.next;
        // 头结点的下一个节点为头结点，从队列链表断掉头结点的引用，帮助GC 回收，Hotspot 默认GC Root可达性分析算法判断对象是否可以回收
        h.next = h; // help GC
        // 头结点换成老二
        head = first;
        // 取出对应的元素
        E x = first.item;
        // 将头结点的元素置为null，因为是带头链表，头结点只是一个哨兵结点，并不存实际元素
        first.item = null;
        // 返回对应元素值
        return x;
    }

链表实现分为带头链表，即头结点是哨兵结点，不存元素，用哨兵结点实现链表，代码会简单很多。

但是 Java 的 [LinkedList][]（LinkedList 源码分析）实现是一个不带头链表，实现代码复杂些，需要处理头结点和尾节点元素逻辑。

LinkedList 是 Josh Bloch 早期实现的，LinkedBlockingQueue 是并发包下的源码，Doug Lea 实现的。

###  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();
                if (c > 1)
                    notEmpty.signal();
            }
        } finally {
            takeLock.unlock();
        }
        if (c == capacity)
            signalNotFull();
        return x;
    }

# 四 总结 #

1、LinkedBlockingQueue 无界阻塞队列，基于单向链表实现，先进先出（FIFO）特性；

2、队列的应用范围很广泛，比如线程池，因为资源有限，线程创建有限，不断重复线程创建开销很大，

通过 “生产-消费” 模型保证线程池核心线程 worker 不退出，重复利用线程，不断执行任务，当然了，

核心线程也是可以设置参数退出，使用也得根据实际情况；

3、synchronized 底层也是通过等待队列实现等待通知，唤醒线程获取锁；

[https_github.com_yihonglei_jdk-source-code-reading]: https://github.com/yihonglei/jdk-source-code-reading
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lobF9qeHk_size_16_color_FFFFFF_t_70]: https://img-blog.csdnimg.cn/20200604142733948.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lobF9qeHk=,size_16,color_FFFFFF,t_70
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lobF9qeHk_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200604164935408.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3lobF9qeHk=,size_16,color_FFFFFF,t_70
[Link 1]: https://blog.csdn.net/yhl_jxy/article/details/86677049
[ReentrantLock]: https://blog.csdn.net/yhl_jxy/article/details/87088314
[LinkedList]: https://blog.csdn.net/yhl_jxy/article/details/109435496