死磕 java集合之LinkedBlockingQueue源码分析

绝地灬酷狼 2022-04-24 09:40 220阅读 0赞

?欢迎关注我的公众号“彤哥读源码”，查看更多源码系列文章, 与彤哥一起畅游源码的海洋。

（手机横屏看源码更方便）

--------------------

## 问题 ##

（1）LinkedBlockingQueue的实现方式？

（2）LinkedBlockingQueue是有界的还是无界的队列？

（3）LinkedBlockingQueue相比ArrayBlockingQueue有什么改进？

## 简介 ##

LinkedBlockingQueue是java并发包下一个以单链表实现的阻塞队列，它是线程安全的，至于它是不是有界的，请看下面的分析。

## 源码分析 ##

### 主要属性 ###

// 容量
    private final int capacity;
    
    // 元素数量
    private final AtomicInteger count = new AtomicInteger();
    
    // 链表头
    transient Node<E> head;
    
    // 链表尾
    private transient Node<E> last;
    
    // take锁
    private final ReentrantLock takeLock = new ReentrantLock();
    
    // notEmpty条件
    // 当队列无元素时，take锁会阻塞在notEmpty条件上，等待其它线程唤醒
    private final Condition notEmpty = takeLock.newCondition();
    
    // 放锁
    private final ReentrantLock putLock = new ReentrantLock();
    
    // notFull条件
    // 当队列满了时，put锁会会阻塞在notFull上，等待其它线程唤醒
    private final Condition notFull = putLock.newCondition();

（1）capacity，有容量，可以理解为LinkedBlockingQueue是有界队列

（2）head, last，链表头、链表尾指针

（3）takeLock，notEmpty，take锁及其对应的条件

（4）putLock, notFull，put锁及其对应的条件

（5）入队、出队使用两个不同的锁控制，锁分离，提高效率

### 内部类 ###

static class Node<E> { 
        E item;
    
        Node<E> next;
    
        Node(E x) {  item = x; }
    }

典型的单链表结构。

### 主要构造方法 ###

public LinkedBlockingQueue() { 
        // 如果没传容量，就使用最大int值初始化其容量
        this(Integer.MAX_VALUE);
    }
    
    public LinkedBlockingQueue(int capacity) { 
        if (capacity <= 0) throw new IllegalArgumentException();
        this.capacity = capacity;
        // 初始化head和last指针为空值节点
        last = head = new Node<E>(null);
    }

### 入队 ###

入队同样有四个方法，我们这里只分析最重要的一个，put(E e)方法：

public void put(E e) throws InterruptedException { 
        // 不允许null元素
        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;
        // 使用put锁加锁
        putLock.lockInterruptibly();
        try { 
            // 如果队列满了，就阻塞在notFull条件上
            // 等待被其它线程唤醒
            while (count.get() == capacity) { 
                notFull.await();
            }
            // 队列不满了，就入队
            enqueue(node);
            // 队列长度加1
            c = count.getAndIncrement();
            // 如果现队列长度如果小于容量
            // 就再唤醒一个阻塞在notFull条件上的线程
            // 这里为啥要唤醒一下呢？
            // 因为可能有很多线程阻塞在notFull这个条件上的
            // 而取元素时只有取之前队列是满的才会唤醒notFull
            // 为什么队列满的才唤醒notFull呢？
            // 因为唤醒是需要加putLock的，这是为了减少锁的次数
            // 所以，这里索性在放完元素就检测一下，未满就唤醒其它notFull上的线程
            // 说白了，这也是锁分离带来的代价
            if (c + 1 < capacity)
                notFull.signal();
        } finally { 
            // 释放锁
            putLock.unlock();
        }
        // 如果原队列长度为0，现在加了一个元素后立即唤醒notEmpty条件
        if (c == 0)
            signalNotEmpty();
    }
    
    private void enqueue(Node<E> node) { 
        // 直接加到last后面
        last = last.next = node;
    }    
    
    private void signalNotEmpty() { 
        final ReentrantLock takeLock = this.takeLock;
        // 加take锁
        takeLock.lock();
        try { 
            // 唤醒notEmpty条件
            notEmpty.signal();
        } finally { 
            // 解锁
            takeLock.unlock();
        }
    }

（1）使用putLock加锁；

（2）如果队列满了就阻塞在notFull条件上；

（3）否则就入队；

（4）如果入队后元素数量小于容量，唤醒其它阻塞在notFull条件上的线程；

（5）释放锁；

（6）如果放元素之前队列长度为0，就唤醒notEmpty条件；

### 出队 ###

出队同样也有四个方法，我们这里只分析最重要的那一个，take()方法：

public E take() throws InterruptedException { 
        E x;
        int c = -1;
        final AtomicInteger count = this.count;
        final ReentrantLock takeLock = this.takeLock;
        // 使用takeLock加锁
        takeLock.lockInterruptibly();
        try { 
            // 如果队列无元素，则阻塞在notEmpty条件上
            while (count.get() == 0) { 
                notEmpty.await();
            }
            // 否则，出队
            x = dequeue();
            // 获取出队前队列的长度
            c = count.getAndDecrement();
            // 如果取之前队列长度大于1，则唤醒notEmpty
            if (c > 1)
                notEmpty.signal();
        } finally { 
            // 释放锁
            takeLock.unlock();
        }
        // 如果取之前队列长度等于容量
        // 则唤醒notFull
        if (c == capacity)
            signalNotFull();
        return x;
    }
    
    private E dequeue() { 
        // head节点本身是不存储任何元素的
        // 这里把head删除，并把head下一个节点作为新的值
        // 并把其值置空，返回原来的值
        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
            notFull.signal();
        } finally { 
            putLock.unlock();
        }
    }

（1）使用takeLock加锁；

（2）如果队列空了就阻塞在notEmpty条件上；

（3）否则就出队；

（4）如果出队前元素数量大于1，唤醒其它阻塞在notEmpty条件上的线程；

（5）释放锁；

（6）如果取元素之前队列长度等于容量，就唤醒notFull条件；

## 总结 ##

（1）LinkedBlockingQueue采用单链表的形式实现；

（2）LinkedBlockingQueue采用两把锁的锁分离技术实现入队出队互不阻塞；

（3）LinkedBlockingQueue是有界队列，不传入容量时默认为最大int值；

## 彩蛋 ##

（1）LinkedBlockingQueue与ArrayBlockingQueue对比？

a）后者入队出队采用一把锁，导致入队出队相互阻塞，效率低下；

b）前才入队出队采用两把锁，入队出队互不干扰，效率较高；

c）二者都是有界队列，如果长度相等且出队速度跟不上入队速度，都会导致大量线程阻塞；

d）前者如果初始化不传入初始容量，则使用最大int值，如果出队速度跟不上入队速度，会导致队列特别长，占用大量内存；

--------------------

欢迎关注我的公众号“彤哥读源码”，查看更多源码系列文章, 与彤哥一起畅游源码的海洋。

![qrcode][]

[qrcode]: /images/20220216/b439f1e379ca49349a6fcaf20529988e.png

发表评论取消回复

表情：

评论列表（有 0 条评论，220人围观）

还没有评论，来说两句吧...

相关阅读

相关死磕 java集合之LinkedBlockingQueue源码分析

?欢迎关注我的公众号“彤哥读源码”，查看更多源码系列文章, 与彤哥一起畅游源码的海洋。（手机横屏看源码更方便） -------------------- 问题（

系统管理员/ 2022年10月01日 05:48/ 0 赞/ 177 阅读

相关死磕 java集合之LinkedBlockingQueue源码分析

?欢迎关注我的公众号“彤哥读源码”，查看更多源码系列文章, 与彤哥一起畅游源码的海洋。（手机横屏看源码更方便） -------------------- 问题（

绝地灬酷狼/ 2022年04月24日 09:40/ 0 赞/ 221 阅读

相关死磕 java集合之ArrayList源码分析

欢迎关注我的公众号“彤哥读源码”，查看更多源码系列文章, 与彤哥一起畅游源码的海洋。简介 ArrayList是一种以数组实现的List，与数组相比，它具有动态扩展的能

梦里梦外;/ 2022年02月26日 09:54/ 0 赞/ 307 阅读

相关死磕 java集合之HashMap源码分析

欢迎关注我的公众号“彤哥读源码”，查看更多源码系列文章, 与彤哥一起畅游源码的海洋。简介 HashMap采用key/value存储结构，每个key对应唯一的value

谁践踏了优雅/ 2022年02月25日 04:36/ 0 赞/ 297 阅读

相关死磕 java集合之WeakHashMap源码分析

欢迎关注我的公众号“彤哥读源码”，查看更多源码系列文章, 与彤哥一起畅游源码的海洋。简介 WeakHashMap是一种弱引用map，内部的key会存储为弱引用，当jv

痛定思痛。/ 2022年02月24日 20:10/ 0 赞/ 217 阅读

相关死磕 java集合之TreeSet源码分析

问题（1）TreeSet真的是使用TreeMap来存储元素的吗？（2）TreeSet是有序的吗？（3）TreeSet和LinkedHashSet有何不同？

亦凉/ 2022年02月19日 15:49/ 0 赞/ 259 阅读

相关死磕 java集合之LinkedHashSet源码分析

问题（1）LinkedHashSet的底层使用什么存储元素？（2）LinkedHashSet与HashSet有什么不同？（3）LinkedHashSet是有序的

忘是亡心i/ 2022年02月19日 06:17/ 0 赞/ 291 阅读

相关死磕 java集合之ConcurrentLinkedQueue源码分析

问题（1）ConcurrentLinkedQueue是阻塞队列吗？（2）ConcurrentLinkedQueue如何保证并发安全？（3）ConcurrentL

落日映苍穹つ/ 2022年02月12日 20:21/ 0 赞/ 325 阅读

相关死磕 java集合之DelayQueue源码分析

问题（1）DelayQueue是阻塞队列吗？（2）DelayQueue的实现方式？（3）DelayQueue主要用于什么场景？简介 DelayQueu

- 日理万妓/ 2022年02月12日 12:34/ 0 赞/ 295 阅读

相关死磕 java集合之LinkedList源码分析

问题（1）LinkedList只是一个List吗？（2）LinkedList还有其它什么特性吗？（3）LinkedList为啥经常拿出来跟ArrayList比较

朴灿烈づ我的快乐病毒、/ 2022年02月09日 12:59/ 0 赞/ 367 阅读