阻塞队列详解 LinkedBlockingQueue 源码分析 -- 基于JDK 1.8

Love The Way You Lie 2021-11-10 00:32 361阅读 0赞

在上一篇我们介绍了[ArrayBlockingQueue 源码分析 -- 基于JDK 1.8][ArrayBlockingQueue _ -- _JDK 1.8]后，

还有这个兄弟阻塞队列类，也顺便一起分析了吧。跟上一篇有很多相似的代码地方，

我们在这里没有重复叙述，所以可以先看一下ArrayBlockingQueue的介绍。

**目录**

1、LinkedBlockingQueue继承结构

2、类中的属性

3、插入元素方法

（1）非阻塞add（offer）方法

（2）、阻塞添加put方法

4、取元素方法

（1）、非阻塞移除头元素方法poll

（2）、非阻塞移除指定元素方法remove

（3）、阻塞take方法，取并移除头元素

（4）、返回队列第一个元素方法peek（不删除）

5、与ArrayBlockingQueue区别

（1）、锁的实现不同

（2）、数据结构不同

（3）、队列大小初始化方式不同

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

# 1、LinkedBlockingQueue继承结构 #

从结构上看LinkedBlockingQueue与ArrayBlockingQueue是一样的；

在IDEA工具中，先找到类源码，使用Ctrl+Shift+Alt+U 快捷键

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xlanVzdGRvaXQ_size_16_color_FFFFFF_t_70][]

# 2、类中的属性 #

![watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xlanVzdGRvaXQ_size_16_color_FFFFFF_t_70 1][]

/**LinkedBlockingQueue从类的名字上看，采用了链表结构，
    
    而Node类就是构成链表的对象节点，从next引用看出这是一个单向链表*/
    static class Node<E> {
        E item;
        Node<E> next;
        Node(E x) { item = x; }
    }
    
    /**初始容量，不指定的话将是Integer.MAX_VALUE*/
    private final int capacity;
    
    /** 元素的个数，在之前ArrayBlockingQueue中的count，
    没有采用AtomicInteger，是因为ArrayBlockingQueue的存取
    全部都是加锁的，包括size方法获取队列大小count也加锁，
    所以非原子变量的count在ArrayBlockingQueue中也是线程安全的；
    
    但是这里改成原子变量了，LinkedBlockingQueue采用了两把锁，
    进队锁和出队锁，在下面的变量中可以看见；
    
    所以支持同时往一个队列里面存元素与取元素，如果同时存取的话，
    count值是线程不安全的，这里设置成原子类型以保证count
    这个临界资源的线程安全*/
    private final AtomicInteger count = new AtomicInteger();
    
    /**头结点，取数据的位置*/
    transient Node<E> head;
    
    /**尾节点，插入数据的位置/
    private transient Node<E> last;
    
    /** 调用方法take, poll等时候的锁*/
    private final ReentrantLock takeLock = new ReentrantLock();
    
    /**等待获取条件，存数据后唤醒取线程，取空以后阻塞等待，
     Wait queue for waiting takes */
    private final Condition notEmpty = takeLock.newCondition();
    
    /** 调用方法 put, offer 等时候获取的锁 */
    private final ReentrantLock putLock = new ReentrantLock();
    
    /** 等待存放条件 ，取数据后唤醒存线程，存满以后阻塞等待，
    Wait queue for waiting puts */
    private final Condition notFull = putLock.newCondition();

# 3、插入元素方法 #

## （1）非阻塞add（offer）方法 ##

在LinkedBlockingQueue类中，没有重写add方法，所以，如果调用add方法，是在父类AbstractQueue中，父类主要是调用offer方法，我们来看LinkedBlockingQueue中的offer方法：

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 {
    
            //队列未满，入队
            if (count.get() < capacity) {
    
                //调用enquene方法，插入元素
                enqueue(node);
    
                //原子自增1
                c = count.getAndIncrement();
    
                //添加元素以后，未满，唤醒其它线程继续添加
                if (c + 1 < capacity)
    
                 //添加后唤醒取元素阻塞的线程
                 notFull.signal();
            }
        } finally {
            putLock.unlock();
        }
    
        //队列空，唤醒其它生产者添加数据
        if (c == 0)
            signalNotEmpty();
        return c >= 0;
    }
    
    
    
    private void signalNotEmpty() {
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lock();
        try {
            notEmpty.signal();
        } finally {
            takeLock.unlock();
        }
    }

同样，有带超时的offer添加方法

public boolean offer(E e, long timeout, TimeUnit unit)

## （2）、阻塞添加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 {
            /*队列满了，则取线程阻塞等待*/
            while (count.get() == capacity) {
                notFull.await();
            }
    
            //否则入队
            enqueue(node);
            c = count.getAndIncrement();
            if (c + 1 < capacity)
                notFull.signal();
        } finally {
            putLock.unlock();
        }
        if (c == 0)
            signalNotEmpty();
    }

# 4、取元素方法 #

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

## （2）、非阻塞移除指定元素方法remove ##

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

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

## （4）、返回队列第一个元素方法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、与ArrayBlockingQueue区别 #

## （1）、锁的实现不同 ##

ArrayBlockingQueue全局只有一个锁，即生产和消费用的是同一个锁；

LinkedBlockingQueue队列中的锁是分离的，即生产用的是putLock，消费是takeLock；所以LinkedBlockingQueue支持同时的并发存、取元素，这方面并发性能优于ArrayBlockingQueue；

## （2）、数据结构不同 ##

ArrayBlockingQueue队列中在生产和消费的时候，是直接将对象在数组中插入或移除的；

LinkedBlockingQueue基于链表队列，在生产和消费的时候，需要把对象转换为Node<E>对象，然后Node<E>对象形成一个单项链表，对链表进行队尾插入或队头移除，这方面操作会多一些，会生成一个额外的Node对象，这在长时间内需要高效并发地处理大批量数据的系统中，会对于GC产生影响。

## （3）、队列大小初始化方式不同 ##

ArrayBlockingQueue队列中必须指定初始化大小；

LinkedBlockingQueue队列中可以不指定队列的大小，默认就会是Integer.MAX\_VALUE，但是这样可能会有问题，当消费者慢于生产者的时候，长期堆积的对象元素，可能会造成内存溢出，所以我们最好还是指定初始化大小；

[ 非阻塞队列详解 ConcurrentLinkedQueue详解 源码分析 -- 基于JDK 1.8][ConcurrentLinkedQueue_ _ -- _JDK 1.8]

[ArrayBlockingQueue _ -- _JDK 1.8]: https://blog.csdn.net/lejustdoit/article/details/97963581
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xlanVzdGRvaXQ_size_16_color_FFFFFF_t_70]: /images/20211109/9dda3943158845ac8e3c38050b3f62f1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2xlanVzdGRvaXQ_size_16_color_FFFFFF_t_70 1]: /images/20211109/6b6fb9d7519f4ed6a8cfd6087d909fa5.png
[ConcurrentLinkedQueue_ _ -- _JDK 1.8]: https://blog.csdn.net/lejustdoit/article/details/98477819