LinkedBlockingQueue源码简析

拼搏现实的明天。 2023-02-15 12:53 157阅读 0赞

上一篇博文画图解析了一下ConcurrentLinkedQueue，那个Queue其实是一个无界队列，按道理可以挂无数个节点在链表中。今天要讲的LinkedBlockingQueue则是一个有界队列，通过他的构造函数来看一下

// 无参的构造函数，默认的队列容量是Integer.MAX_VALUE，基本可以认为是一个无界队列了
    public LinkedBlockingQueue() { 
       this(Integer.MAX_VALUE);
    }
    
    // 最常用的构造函数，传入capacity作为队列的容量
    public LinkedBlockingQueue(int capacity) { 
        if (capacity <= 0) throw new IllegalArgumentException();
        this.capacity = capacity;
        last = head = new Node<E>(null);
    }

源码在定义变量的时候，有下面这几个变量，大家应该也都不陌生了，分别是两个ReentrantLock 和两个Condition

/** Lock held by take, poll, etc */
    private final ReentrantLock takeLock = new ReentrantLock();
    
    /** Wait queue for waiting takes */
    private final Condition notEmpty = takeLock.newCondition();
    
    /** Lock held by put, offer, etc */
    private final ReentrantLock putLock = new ReentrantLock();
    
    /** Wait queue for waiting puts */
    private final Condition notFull = putLock.newCondition();

先通过一张图来简单描述一下LinkedBlockingQueue的原理，之后再来分析里面的源码  
![LinkedBlockingQueue原理图][LinkedBlockingQueue]  
当有线程在进行入队操作的时候，需要先获取到putLock，这样其他的线程就无法进行入队操作了，保证了线程安全，而notFull这个Condition在这的作用就是当队列的容量已满的时候，触发Condition，不再让线程继续往队列中塞数据了，入队的线程阻塞住，直到有其他的线程取出了LinkedBlockingQueue的数据，入队的线程才会被唤醒

在进行出队操作的时候，线程则是获取takeLock，不断从队列中取出数据，当LinkedBlockingQueue中的数据全部出队了之后，再来线程出队的话会被notEmpty阻塞住，直到有其他的线程进行了入队操作

知道了大概的原理之后，我们再来通过源码看看具体是如何实现的吧  
先来看入队的操作

public void put(E e) throws InterruptedException { 
        if (e == null) throw new NullPointerException();
        int c = -1;
        // 需要入队的node节点
        Node<E> node = new Node<E>(e);
        final ReentrantLock putLock = this.putLock;
        // 队列中的元素数量
        final AtomicInteger count = this.count;
        // 加锁保证线程安全
        // 阻塞式的获取锁，如果加锁失败，则会被挂起；加锁过程被打断会报错
        putLock.lockInterruptibly();
        try { 
    		// 当LinkedBlockingQueue中的元素数量等于最大容量时
    		// 触发notFull这个Condition，当前线程释放锁，加入Condition等待队列，不允许这个线程再继续往队列中加入更多的元素了
            while (count.get() == capacity) { 
                notFull.await();
            }
            // node入队，在队尾加入新的node，这个逻辑主要是设计到了一些指针的变换，所以就不展开细讲了
            enqueue(node);
            // 先获取count的值赋给c，然后再加1
            c = count.getAndIncrement();
            // 如果c + 1 < capacity满足的话，说明count值也是小于capacity，即队列中的元素还没有满，那么就唤醒被notFull阻塞的线程
            if (c + 1 < capacity)
                notFull.signal();
        } finally { 
            putLock.unlock();
        }
        // 这里的c == 0满足的话，那么count值就是1，也就是说队列中存在了一个数据，那么就唤醒NotEmpty，让被阻塞的出队线程继续出队
        if (c == 0)
            signalNotEmpty();
    }

看完了入队，那么再来分析一下出队的源码

public E take() throws InterruptedException { 
        E x;
        int c = -1;
        final AtomicInteger count = this.count;
        final ReentrantLock takeLock = this.takeLock;
        // 阻塞式加锁
        takeLock.lockInterruptibly();
        try { 
        	// 如果队列中元素的数量等于0的时候，则使用notEmpty阻塞线程，当前线程释放锁，加入Condition等待队列
        	// 线程无法再进行出队的操作了
            while (count.get() == 0) { 
                notEmpty.await();
            }
            // 这个操作是将队头的元素出队，也是涉及到了指针的各种变换
            x = dequeue();
            c = count.getAndDecrement();
            // c > 1则说明count > 0，队列中有数据的话，则唤醒被notEmpty阻塞的线程，继续进行出队的操作
            if (c > 1)
                notEmpty.signal();
        } finally { 
            takeLock.unlock();
        }
        // 当c == capacity，那么count = capacity - 1，即队列还没有达到饱和的状态，唤醒入队的线程继续入队 
        if (c == capacity)
            signalNotFull();
        return x;
    }

看完了出队和入队的这两个方法的源码，我们发现其实出队和入队的操作都是大同小异的，使用ReentrantLock来进行加锁，保证了线程安全，同时使用Condition来实现了阻塞的效果，也就是LinkedBlockingQueue名称中的Blocking的效果

最后再稍微来看看LinkedBlockingQueue的Iterator方法，我这里只贴出来部分的代码

new一个Iterator的逻辑

Itr() { 
        fullyLock();
        try { 
            current = head.next;
            if (current != null)
                currentElement = current.item;
        } finally { 
            fullyUnlock();
        }
    }

获取下一个元素的next方法

public E next() { 
        fullyLock();
        try { 
            if (current == null)
                throw new NoSuchElementException();
            E x = currentElement;
            lastRet = current;
            current = nextNode(current);
            currentElement = (current == null) ? null : current.item;
            return x;
        } finally { 
            fullyUnlock();
        }
    }

这里我们不关注里面一些具体的实现逻辑，我们关注的点应该是在fullyLock()这个方法，我们发现在new一个Iterator和next()方法中都使用到了fullyLock()，那么这个方法是实现了什么呢

/** * Locks to prevent both puts and takes. */
    void fullyLock() { 
        putLock.lock();
        takeLock.lock();
    }

通过源码，很简单就能发现其实就是将我们上面说的putLock和takeLock都上了锁，所以说在进行Iterator遍历操作的时候，是无法再进行出队和入队的操作的，这两把锁都被Iterator给锁住了  
而出队和入队分别是使用的putLock和takeLock，所以出队和入队其实是可以并发执行的，不会互相影响

LinkedBlockingQueue的实现原理也挺简单的，其实就是两把锁实现线程安全，同时出队和入队操作分别使用了各自的锁，提高了并发效率，只有在进行Iterator操作的时候，会阻塞住所有的入队和出队操作，而阻塞的效果则是由两个Condition来实现

[LinkedBlockingQueue]: https://img-blog.csdnimg.cn/20200619154651471.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L0xPX1lVTg==,size_16,color_FFFFFF,t_70