LinkedBlockingQueue队列源码解析

桃扇骨 2023-06-05 08:15 8阅读 0赞

#### 前言： ####

1.  上一篇我们讲解了亲戚：ArrayBlockingQueue ： [https://blog.csdn.net/zhangyong01245/article/details/102390093][https_blog.csdn.net_zhangyong01245_article_details_102390093] 的相关知识，今天我们了解一下 LinkedBlockingQueue
2.  ArrayBlockingQueue 和 LinkedBlockingQueue 相当于 ArrayList 和 LinkedList 一样 ，LinkedBlockingQueue 也是由链表来存储数据的。
3.  LinkedBlockingQueue 如果不指定容量，默认为：**Integer.MAX\_VALUE**，也就是无界队列，当入队 > 出队 可能会导致机器内存溢出。 建议设置存储大小

#### LinkedBlockingQueue源码中参数定义： ####

// 储存数据的节点
        static class Node<E> {
            E item;
    
            // 下一个节点指向
            Node<E> next;
    
            Node(E x) { item = x; }
        }
    
        // 链表的容量， 默认是 Integer.MAX_VALUE
        private final int capacity;
    
        // 当前的链表大小
        private final AtomicInteger count = new AtomicInteger();
    
        // 头结点
        transient Node<E> head;
    
        // 尾节点
        private transient Node<E> last;
    
        // 获取并移除元素所使用的的锁
        private final ReentrantLock takeLock = new ReentrantLock();
    
        // 链表不为空 Condition
        private final Condition notEmpty = takeLock.newCondition();
    
        // 添加元素时使用的锁
        private final ReentrantLock putLock = new ReentrantLock();
    
        // 连表没有达到最大容量的 Condition
        private final Condition notFull = putLock.newCondition();

##### 构造方法的定义： #####

1.  LinkedBlockingQueue（）： 无参构造方法， 链表的大小默认为 Integer.MAX\_VALUE ， 并初始化 链表的 head 和 last
2.  LinkedBlockingQueue(int capacity) ： 包含 int 的构造方法，链表的大小 capacity，并初始化 链表的 head 和 last
3.  LinkedBlockingQueue(Collection<? extends E> c) ：包含 Collection 集合的构造方法，初始化链表 大小默认为 Integer.MAX\_VALUE，并将集合中的参数放入到 链表中：
4.  源码：

/**
         * 1. 初始化
         * 2. 链表的大小默认为 Integer.MAX_VALUE
         */
        public LinkedBlockingQueue() {
            this(Integer.MAX_VALUE);
        }
    
    
        /**
         * 1. 初始化
         * 2. 链表的大小默认为 capacity
         * 3. 初始化 头、尾节点
         */
        public LinkedBlockingQueue(int capacity) {
            if (capacity <= 0) throw new IllegalArgumentException();
            this.capacity = capacity;
            last = head = new Node<E>(null);
        }
    
        /**
         *  1. 初始化
         *  2. 链表的大小默认为 Integer.MAX_VALUE
         *  3. 将集合 c 中的元素依次放入到链表中
         */
        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();
                    if (n == capacity)
                        throw new IllegalStateException("Queue full");
                    enqueue(new Node<E>(e));
                    ++n;
                }
                count.set(n);
            } finally {
                putLock.unlock();
            }
        }

#### 入队方法 ####

1.  enqueue(Node node)： 私有方法，提供给 put、offer 调用。
2.  put(E e) : 入队方法，队列已满，阻塞等待，while 循环等待直到队列未满时，在进行入队操作。
3.  offer(E e)：入队方法，队列已满时直接返回Flase。
4.  offer(E e, long timeout, TimeUnit unit)： 包含时间参数的offer 方法，在等待指定时间后，后续执行操作和 offer(E e) 类似
5.  源码：

/**
         *  1、入队一个元素
         *  2、私有的方法，仅供内部使用
         */
        private void enqueue(Node<E> node) {
            // assert putLock.isHeldByCurrentThread();
            // assert last.next == null;
            last = last.next = node;
        }
    	
    	/**
         * 1. 入队一个元素
         * 2. 队列已满，阻塞等待
         * 3. 队列未满，新建节点 node，并调用 enqueue 放置到链表中，
         *    放置后队列未满，唤醒其它入队线程；
         */
        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();
        }
    
        /**
         * 1. 入队一个元素
         * 2. 队列已满，直接返回False
         * 3. 队列未满，新建节点 node，并调用 enqueue 放置到链表中，
         *    放置后队列未满，唤醒其它入队线程；
         */
        public boolean offer(E e) {
            if (e == null) throw new NullPointerException();
            final AtomicInteger count = this.count;
            // 队列已满，直接返回False
            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);
                    c = count.getAndIncrement();
                    if (c + 1 < capacity)
                        notFull.signal();
                }
            } finally {
                putLock.unlock();
            }
            // 如果放之前队列中没有元素，放完以后要唤醒出队线程进行出队。
            if (c == 0)
                signalNotEmpty();
            return c >= 0;
        }
    
    	/**
         * 相对于 offer（E e），当队列已满时，在等待指定的时间后，
         * 操作与 offer 相同
         */
        public boolean offer(E e, long timeout, TimeUnit unit)
                throws InterruptedException {
    
            if (e == null) throw new NullPointerException();
            long nanos = unit.toNanos(timeout);
            int c = -1;
            final ReentrantLock putLock = this.putLock;
            final AtomicInteger count = this.count;
            putLock.lockInterruptibly();
            try {
                while (count.get() == capacity) {
                    if (nanos <= 0)
                        return false;
                    nanos = notFull.awaitNanos(nanos);
                }
                enqueue(new Node<E>(e));
                c = count.getAndIncrement();
                if (c + 1 < capacity)
                    notFull.signal();
            } finally {
                putLock.unlock();
            }
            if (c == 0)
                signalNotEmpty();
            return true;
        }
    
    	/**
         * 唤醒 其他出队线程 进行出队
         */
        private void signalNotEmpty() {
            final ReentrantLock takeLock = this.takeLock;
            takeLock.lock();
            try {
                notEmpty.signal();
            } finally {
                takeLock.unlock();
            }
        }

#### 出队方法： ####

1.  dequeue() : 私有出队方法，提供给 take、poll 调用
2.  take()： 队列为空时，**阻塞线程**，直到队列中存在新的元素，调用 dequeue 出队，
3.  poll(): 出队方法 和 take 的区别是，当队列为空时，直接返回 null。
4.  poll(long timeout, TimeUnit unit)： 加入了等待时间程序，当队列为空时，在等待指定的时间后，队列还为空，则返回null， 反之出队列一个元素。
5.  peek()： 返回下一个需要出队的元素，但并不出队
6.  源码：

/**
         * 1. 出队一个元素
         * 2. 私有的方法，仅供内部使用
         */
        private E dequeue() {
            // assert takeLock.isHeldByCurrentThread();
            // assert head.item == null;
            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;
        }
    
    
        /**
         *  1、调用 dequeue 出队一个元素
         *  2、出队后还存在元素唤醒下一个出队线程
         *
         */
        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;
        }
    
        /**
         * 与 take() 方法的不同之处是，如队列中没有元素则直接返回 null
         * @return
         */
        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;
        }
    
        /**
         * 与 无参 poll 方法的不同之处是，若队列中没有元素等待指定的时间后继续进行
         */
        public E poll(long timeout, TimeUnit unit) throws InterruptedException {
            E x = null;
            int c = -1;
            long nanos = unit.toNanos(timeout);
            final AtomicInteger count = this.count;
            final ReentrantLock takeLock = this.takeLock;
            takeLock.lockInterruptibly();
            try {
                while (count.get() == 0) {
                    if (nanos <= 0)
                        return null;
                    nanos = notEmpty.awaitNanos(nanos);
                }
                x = dequeue();
                c = count.getAndDecrement();
                if (c > 1)
                    notEmpty.signal();
            } finally {
                takeLock.unlock();
            }
            if (c == capacity)
                signalNotFull();
            return x;
        }
    
    
        /**
         *  返回下一个需要出队的元素，但并不出队
         */
        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();
            }
        }
    
    	/**
         * 唤醒 其它入队线程 进行添加
         */
        private void signalNotFull() {
            final ReentrantLock putLock = this.putLock;
            putLock.lock();
            try {
                notFull.signal();
            } finally {
                putLock.unlock();
            }
        }

##### 其他方法： #####

1.  fullyLock(): 出队 和 入队 全部加锁
2.  fullyUnlock(): 出队 和 入队 全部解锁
3.  size(): 返回队列中元素的大小
4.  remainingCapacity() : 返回队列的剩余空间
5.  源码：

/**
         * 出队 和 入队 全部加锁
         */
        void fullyLock() {
            putLock.lock();
            takeLock.lock();
        }
    
        /**
         * 出队 和 入队 全部解锁
         */
        void fullyUnlock() {
            takeLock.unlock();
            putLock.unlock();
        }
    
        // 返回队列中元素的大小
        public int size() {
            return count.get();
        }
    
        // 还可以存储多少个元素
        public int remainingCapacity() {
            return capacity - count.get();
        }

#### End： ####

在学习 LinkedBlockingQueue 之前最好是能够 LinkedList 和 ReentrantLock 有一定的了解，然后在查阅其源码时，便会手到擒来。

[https_blog.csdn.net_zhangyong01245_article_details_102390093]: https://blog.csdn.net/zhangyong01245/article/details/102390093