JUC集合类 LinkedBlockingQueue源码解析 JDK8

你的名字 2023-03-01 13:40 13阅读 0赞

### 文章目录 ###

*  前言
 *  成员
 *  构造器
 *  入队
 *   *  add
     *  offer
     *  put
     *  超时offer
     *  入队方法总结
 *  出队
 *   *  remove
     *  poll
     *  take
     *  超时poll
     *  出队方法总结
 *  内部删除 remove(Object o)
 *  获取操作
 *   *  peek
     *  element
 *  迭代器
 *  总结

# 前言 #

LinkedBlockingQueue是一种FIFO（first-in-first-out 先入先出）的有界阻塞队列，底层是单链表，也支持从内部删除元素。并发操作依赖于加锁的控制，支持阻塞式的入队出队操作。

相比ArrayBlockingQueue的一个Lock，LinkedBlockingQueue使用了两个Lock，分别对应入队动作和出队动作，这便提高了并发量。

[JUC框架 系列文章目录][JUC_]

# 成员 #

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

LinkedBlockingQueue的底层实现基于单链表，上面是单链表的Node定义。

/** 容量，毕竟这是一个有界队列 */
        private final int capacity;
    
        /** 大小，元素的个数 */
        private final AtomicInteger count = new AtomicInteger();
    
        //队首指针
        transient Node<E> head;
    
        //队尾指针
        private transient Node<E> last;

上面就是些基于单链表的队列的必备成员。之所以需要使用AtomicInteger，是因为有两个线程（入队线程、出队线程）可能同时在修改它，所以用原子类来保持count的正确性。

/** 出队线程需要竞争这把锁，竞争到了才能出队，也就是说同时只有一个线程能出队 */
        private final ReentrantLock takeLock = new ReentrantLock();
    
        /** 出队线程可能会暂时阻塞在这个AQS条件队列里，当发现队列已空时 */
        private final Condition notEmpty = takeLock.newCondition();
    
        /** 入队线程需要竞争这把锁，竞争到了才能入队，也就是说同时只有一个线程能入队 */
        private final ReentrantLock putLock = new ReentrantLock();
    
        /** 入队线程可能会暂时阻塞在这个AQS条件队列里，当发现队列已满时 */
        private final Condition notFull = putLock.newCondition();

为了保证last的正确性，只有竞争到putLock的入队线程才能执行入队动作。这样就只有一个线程在修改last。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FubGlhbjUyMw_size_16_color_FFFFFF_t_70]

上图展示了入队线程的通用流程。当入队线程从`notFull.await()`处恢复执行时，已经又重新获得了putLock，然后入队线程即将执行入队动作，别的线程也不可能和它竞争入队了。

为了保证head的正确性，只有竞争到takeLock的出队线程才能执行出队动作。这样就只有一个线程在修改head。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FubGlhbjUyMw_size_16_color_FFFFFF_t_70 1]  
上图展示了出队线程的通用流程。当出队线程从`notEmpty.await()`处恢复执行时，已经又重新获得了takeLock，然后出队线程即将执行出队动作，别的线程也不可能和它竞争出队了。

# 构造器 #

public LinkedBlockingQueue() { 
            this(Integer.MAX_VALUE);//默认大小
        }
    
        public LinkedBlockingQueue(int capacity) { 
            if (capacity <= 0) throw new IllegalArgumentException();
            this.capacity = capacity;
            last = head = new Node<E>(null);//队列始终有一个dummy node作为head
        }

默认大小为Integer.MAX\_VALUE，当然这也是最大大小。队列始终有一个dummy node作为head。

# 入队 #

## add ##

//AbstractQueue.java
        public boolean add(E e) { 
            if (offer(e))
                return true;
            else//返回false的处理不一样
                throw new IllegalStateException("Queue full");
        }
        
    //Queue.java(接口文件)
    	boolean offer(E e);

这个方法在LinkedBlockingQueue.java中找不到，因为你直接调用的是父类实现。`add`依靠于子类的`offer`实现。所以，`add`就是在调用自己的`offer`方法，只不过有点绕。

## offer ##

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 { 
                //入队前获取最新count，来判断
                if (count.get() < capacity) { //只进行一次尝试
                    enqueue(node);
                    c = count.getAndIncrement();//执行count++
                    if (c + 1 < capacity)//如果新大小 小于容量
                        notFull.signal();//让一个入队线程离开AQS条件队列
                }
            } finally { 
                putLock.unlock();
            }
            if (c == 0)//如果新大小为1
                signalNotEmpty();//让一个出队线程离开AQS条件队列
            return c >= 0;//如果新大小>=1，说明入队成功
        }

该函数只进行一次尝试，如果队列当前已满，就直接退出；如果队列当前非满，才执行入队动作。它甚至会在获得锁之前，就判断队列满的情况。

`if (c + 1 < capacity)`和`if (c == 0)`两处用的比较运算符不一样：

*  `if (c + 1 < capacity)`处。因为当前线程正持有putLock中（所以count不可能被别的线程增加），但count可能由于别的出队线程而减小，所以只要新size小于capacity，就唤醒后面的入队线程。
 *  `if (c == 0)`处。当旧size为0时，才去唤醒后面的出队线程。因为旧size为正数的话，出队线程是不会阻塞的，所以只需要精确判断这种情况。

## put ##

1.  `putLock.lockInterruptibly()`中：因获得不到锁而阻塞，是可以被中断而抛出中断异常。
2.  try代码块中：如果没有中断来临，该函数会一直阻塞直到它成功入队。如果队列一直是满的，我们可以通过中断线程来终止`put`的调用。

public void put(E e) throws InterruptedException { 
            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;
            putLock.lockInterruptibly();//可能抛出中断异常
            try { 
                while (count.get() == capacity) { //从下一句恢复后，需要检查是否为虚假唤醒
                    notFull.await();//可能抛出中断异常
                }
                //执行到这里，count肯定只会比capacity小
                enqueue(node);
                c = count.getAndIncrement();//执行count++
                if (c + 1 < capacity)
                    notFull.signal();
            } finally { 
                putLock.unlock();
            }
            if (c == 0)
                signalNotEmpty();
        }

这里需要讲一下count的正确性，从`while (count.get() == capacity)`退出时count肯定小于capacity了，并且我们不用担心接下来的并发问题：

*  不用担心另一种线程——出队线程。因为出队线程只会使得capacity变小，所以即使有出队线程的并发，count还是小于capacity的。
 *  不用担心自己线程——入队线程。因为当前线程还拥有着putLock。
 *  综上，从循环退出后，count将保持小于capacity。而这是执行`enqueue(node)`的前提。

private void enqueue(Node<E> node) { 
            // assert putLock.isHeldByCurrentThread();
            // assert last.next == null;
            last = last.next = node;
        }

`enqueue`函数将新节点插到尾部，然后last更新为新节点。

## 超时offer ##

1.  `putLock.lockInterruptibly()`中：因获得不到锁而阻塞，是可以被中断而抛出中断异常。
2.  try代码块中：如果没有中断或超时来临，该函数会一直阻塞直到它成功入队。超时前我们可以通过中断线程来终止`offer`的调用，超时后如果队列还是满的`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)//如果队列是满的，且剩余等待时间<= 0这代表不用等待，所以直接返回false
                        return false;
                    //下面这句可能抛出中断异常
                    nanos = notFull.awaitNanos(nanos);//返回剩余等待时间，如果超时，返回值小于0
                }
                enqueue(new Node<E>(e));
                c = count.getAndIncrement();
                if (c + 1 < capacity)
                    notFull.signal();
            } finally { 
                putLock.unlock();
            }
            if (c == 0)
                signalNotEmpty();
            return true;
        }

## 入队方法总结 ##

<table> 
 <thead> 
  <tr> 
   <th align="center">入队方法</th> 
   <th align="center">是否等待</th> 
   <th align="center">队列满时的处理</th> 
   <th align="center">返回值</th> 
   <th align="center">返回值含义</th> 
   <th align="center">抛出中断异常的含义</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="center">add</td> 
   <td align="center">一次入队尝试，从不等待</td> 
   <td align="center">抛出"Queue full"异常</td> 
   <td align="center">true<br>-</td> 
   <td align="center">入队成功<br>-</td> 
   <td align="center">-</td> 
  </tr> 
  <tr> 
   <td align="center">offer</td> 
   <td align="center">一次入队尝试，从不等待</td> 
   <td align="center">返回false</td> 
   <td align="center">true<br>false</td> 
   <td align="center">入队成功<br>入队失败</td> 
   <td align="center">-</td> 
  </tr> 
  <tr> 
   <td align="center">put</td> 
   <td align="center">入队尝试失败后，会等待</td> 
   <td align="center">进入条件队列继续等待</td> 
   <td align="center">void</td> 
   <td align="center">只要从<code>put</code>调用处正常返回，就代表入队成功</td> 
   <td align="center">signal来临前，中断发生</td> 
  </tr> 
  <tr> 
   <td align="center">超时offer</td> 
   <td align="center">入队尝试失败后，会等待</td> 
   <td align="center">如果没超时，则进入条件队列继续等待；<br>如果超时了，返回false</td> 
   <td align="center">true<br>false</td> 
   <td align="center">规定时间内，入队成功<br>规定时间内，没有入队</td> 
   <td align="center">signal或超时来临前，中断发生</td> 
  </tr> 
 </tbody> 
</table>

# 出队 #

## remove ##

//AbstractQueue.java
        public E remove() { 
            E x = poll();
            if (x != null)
                return x;
            else
                throw new NoSuchElementException();
        }
    
        
    //Queue.java(接口文件)
        E poll();

同样的，`remove`就是在调用自己的`poll`方法。

## poll ##

public E poll() { 
            final AtomicInteger count = this.count;
            if (count.get() == 0)//提前进行一个快捷判断，发现队列已空，则退出
                return null;//返回null，代表出队失败
            E x = null;
            int c = -1;
            final ReentrantLock takeLock = this.takeLock;
            takeLock.lock();
            try { 
                //出队前获取最新count，来判断
                if (count.get() > 0) { //只进行一次尝试
                    x = dequeue();
                    c = count.getAndDecrement();
                    if (c > 1)//如果旧大小 大于等于2
                        notEmpty.signal();//让一个出队线程离开AQS条件队列
                }
            } finally { 
                takeLock.unlock();
            }
            if (c == capacity)//如果旧大小 等于 容量
                signalNotFull();//才需要让一个入队线程离开AQS条件队列
            return x;//返回非null
        }

该函数只进行一次尝试，如果队列当前已空，就直接退出；如果队列当前非空，才执行出队动作。它甚至会在获得锁之前，就判断队列空的情况。

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

`dequeue`函数很简单：

*  让head后移一个节点。
 *  让移除掉的旧head的next指针指向自己，以区别于队尾节点（队尾节点的next为null）。
 *  让新head变成dummy node之前，保存其item以返回。

## take ##

1.  `takeLock.lockInterruptibly()`中：因获得不到锁而阻塞，是可以被中断而抛出中断异常。
2.  try代码块中：如果没有中断来临，该函数会一直阻塞直到它成功出队。如果队列一直是空的，我们可以通过中断线程来终止`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();//可能抛出中断异常
                }
                //执行到这里，count肯定只会比0大
                x = dequeue();
                c = count.getAndDecrement();
                if (c > 1)
                    notEmpty.signal();
            } finally { 
                takeLock.unlock();
            }
            if (c == capacity)
                signalNotFull();
            return x;
        }

## 超时poll ##

1.  `takeLock.lockInterruptibly()`中：因获得不到锁而阻塞，是可以被中断而抛出中断异常。
2.  try代码块中：如果没有中断或超时来临，该函数会一直阻塞直到它成功出队。超时前我们可以通过中断线程来终止`poll`的调用，超时后如果队列还是空的`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)//如果队列是空的，且剩余等待时间<= 0这代表不用等待，所以直接返回null
                        return null;
                    //下面这句可能抛出中断异常
                    nanos = notEmpty.awaitNanos(nanos);//返回剩余等待时间，如果超时，返回值小于0
                }
                x = dequeue();
                c = count.getAndDecrement();
                if (c > 1)
                    notEmpty.signal();
            } finally { 
                takeLock.unlock();
            }
            if (c == capacity)
                signalNotFull();
            return x;
        }

## 出队方法总结 ##

<table> 
 <thead> 
  <tr> 
   <th align="center">出队方法</th> 
   <th align="center">是否等待</th> 
   <th align="center">队列空时的处理</th> 
   <th align="center">返回值</th> 
   <th align="center">返回值含义</th> 
   <th align="center">抛出中断异常的含义</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td align="center">remove</td> 
   <td align="center">一次出队尝试，从不等待</td> 
   <td align="center">抛出NoSuchElementException</td> 
   <td align="center">非null<br>-</td> 
   <td align="center">队列非空<br>-</td> 
   <td align="center">-</td> 
  </tr> 
  <tr> 
   <td align="center">poll</td> 
   <td align="center">一次出队尝试，从不等待</td> 
   <td align="center">返回null</td> 
   <td align="center">非null<br>null</td> 
   <td align="center">队列非空<br>队列空</td> 
   <td align="center">-</td> 
  </tr> 
  <tr> 
   <td align="center">take</td> 
   <td align="center">出队尝试失败后，会等待</td> 
   <td align="center">进入条件队列继续等待</td> 
   <td align="center">void</td> 
   <td align="center">只要从<code>take</code>调用处返回，就代表出队成功</td> 
   <td align="center">signal来临前，中断发生</td> 
  </tr> 
  <tr> 
   <td align="center">超时poll</td> 
   <td align="center">出队尝试失败后，会等待</td> 
   <td align="center">如果没超时，则进入条件队列继续等待；<br>如果超时了，返回false</td> 
   <td align="center">非null<br>null</td> 
   <td align="center">规定时间内，出队成功<br>规定时间内，没有出队</td> 
   <td align="center">signal或超时来临前，中断发生</td> 
  </tr> 
 </tbody> 
</table>

# 内部删除 remove(Object o) #

public boolean remove(Object o) { 
            if (o == null) return false;
            fullyLock();
            try { 
                //p是循环变量，trail用来保存旧p
                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();
            }
        }

循环遍历找那个元素，如果找到了就删除，这可能是一个内部删除。总之，有可能是队首或队尾，所以需要两个锁都先持有了再做操作，结束后两把锁都释放了。

void fullyLock() { 
        putLock.lock();
        takeLock.lock();
    }
    
    void fullyUnlock() { 
        takeLock.unlock();
        putLock.unlock();
    }

void unlink(Node<E> p, Node<E> trail) { 
            // assert isFullyLocked();
            p.item = null;//逻辑删除
            trail.next = p.next;//将p从链表中移除
            if (last == p)//这种情况需要更新last
                last = trail;
            if (count.getAndDecrement() == capacity)
                notFull.signal();
        }

一个正常的从单链表中删除一个节点的操作。但注意没有将p的next指针指向自己，因为这样可以让迭代器继续从逻辑删除的p节点后继续遍历。

注意，此`unlink`函数不需要去执行`if (c > 1) notEmpty.signal()`的操作，因为能执行这个函数说明队列中至少有一个元素，那么在`fullyLock`之前就不可能有出队线程因为队列为空而阻塞。

# 获取操作 #

## peek ##

public E peek() { 
            if (count.get() == 0)//快捷判断，如果队列空则直接返回null
                return null;
            final ReentrantLock takeLock = this.takeLock;
            takeLock.lock();//先拿到出队锁，以免有人更新head
            try { 
                Node<E> first = head.next;
                if (first == null)
                    return null;
                else
                    return first.item;
            } finally { 
                takeLock.unlock();
            }
        }

## element ##

//AbstractQueue.java
        public E element() { 
            E x = peek();
            if (x != null)
                return x;
            else//队列为空，抛出异常
                throw new NoSuchElementException();
        }

# 迭代器 #

此迭代器是弱一致性的。因为即使节点被删除，迭代器也会照样返回被删除节点的item。

弱一致性是因为并发操作。当迭代器遍历到某个位置后，你调用`hasNext`返回true说明下一个节点存在。但之后有别人删除掉了你的这个节点，然后你再调用`next()`理论上来说我应该返回这个节点的item给你，但删除操作会使得节点的item为null，所以迭代器中必须使用`E currentElement`提前保存。

private class Itr implements Iterator<E> { 
            private Node<E> current;//下一次next()将返回的数据
            private Node<E> lastRet;//上一次next()已返回的数据
            private E currentElement;//下一次next()将返回的数据
    
            Itr() { 
                fullyLock();
                try { 
                    current = head.next;//构造时，就准备好current的两个成员
                    if (current != null)
                        currentElement = current.item;
                } finally { 
                    fullyUnlock();
                }
            }
    
            public boolean hasNext() { 
                return current != null;//只要current不为null，即使它的item变成null了，
                //接下来的next()我们也得返回一个非null值，所以需要用currentElement提前保存
            }
    
            private Node<E> nextNode(Node<E> p) { 
                for (;;) { 
                    Node<E> s = p.next;
                    if (s == p)//如果是已出队节点，则跳转到head后继
                        return head.next;
                    //1. 如果后继s为null。说明p已经是最后一个节点，遍历已到终点，返回null
                    //2. 如果后继s的item不为null。正常节点，返回它即可
                    if (s == null || s.item != null)
                        return s;
                    //执行到这里，说明后继s的item为null，这是一个逻辑删除的节点，
                    //但通过它的后继我们能找到正常节点，所以让p后移，继续循环
                    p = s;
                }
            }
    
            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();
                }
            }
    
            public void remove() { 
                if (lastRet == null)
                    throw new IllegalStateException();
                fullyLock();
                try { 
                    Node<E> node = lastRet;
                    lastRet = null;//让此函数无法连续调两次
                    for (Node<E> trail = head, p = trail.next;//从头遍历，以找到这个节点
                         p != null;
                         trail = p, p = p.next) { 
                        if (p == node) { //找到了同一个对象，才删除它（不是equals判断哦）
                            unlink(p, trail);
                            break;
                        }
                    }
                } finally { 
                    fullyUnlock();
                }
            }
        }

# 总结 #

*  和ConcurrentLinkedQueue一样，初始化时有一个dummy node。也就是说，真正的数据节点，永远是head的后继。
 *  使用了两个`Lock`，分别负责修改`head`和`last`。之所以可以这样，是因为队列非空非满的时候，同时入队出队是互不影响，而且count是一个原子类。
 *  两个`Condition`的使用，是控制阻塞等待的关键。
 *  两个`Lock`都是非公平模式的获取锁方式，抢锁更快，提高并发。
 *  出队的节点next指向自身，以区别于队尾节点（next为null）。

[JUC_]: https://blog.csdn.net/anlian523/article/details/106598910
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FubGlhbjUyMw_size_16_color_FFFFFF_t_70]: /images/20230208/97738bd9e46a4b5383ace009fbc5ff33.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FubGlhbjUyMw_size_16_color_FFFFFF_t_70 1]: https://img-blog.csdnimg.cn/20200727231529529.jpg?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L2FubGlhbjUyMw==,size_16,color_FFFFFF,t_70