本文共 3227 字，大约阅读时间需要 10 分钟。

孤独，所有人都是孤独的，没有人能独自超脱这一切。

——玛娅·安杰格

0 前言

LinkedBlockingQueue - 单链表实现的阻塞队列。该队列按 FIFO（先进先出）排序元素，新元素从队列尾部插入，从队首获取元素.是深入并发编程的基础数据结构.

1 继承体系

• Queue 作为最基础的接口,定义了队列的三大类基本操作：

  

• BlockingQueue 即在 Queue 的基础上加上了阻塞的概念，比如

  • 一直阻塞
  • 阻塞一定时间,返回特殊值

remove 方法，BlockingQueue 类注释中定义的是抛异常，但 LinkedBlockingQueue 中 remove 方法实际是返回 false。

2 属性

2.1 链式存储

• 节点的数据结构
  
• next 为当前元素的下一个，为 null 表示当前节点是最后一个
• 链表容量
  
  默认大小为 Integer.MAX_VALUE ！显然太大了！禁用！
  
• 链表已有元素大小
  使用了 AtomicInteger，线程安全
  
• 链表头、尾
  
  

2.2 锁

• take 时的锁
  
• take 的条件队列，condition 可以简单理解为基于 ASQ 建立的条件队列
  
• put 时的锁
  
• put 的条件队列
  

锁有 take 锁和 put 锁，是为了保证队列操作时的线程安全，设计两种锁，是为了 take 和 put 两种操作可以同时进行，而互不影响。

2.3 迭代器

• 实现了自己的迭代器
  

3 构造方法

3.1 无参

• 默认为 Integer 的最大值
  

3.2 有参

• 指定容量大小.链表头尾相等，节点值（item）都是 null

  

• 已有集合数据

public LinkedBlockingQueue(Collection
    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) {
               // 集合元素不能为 null            if (e == null)                throw new NullPointerException();            // capacity 代表链表的大小，在这为 Integer.MAX_VALUE            // 若集合大小 > IntegerMAX_VALUE,直接抛异常.            if (n == capacity)                throw new IllegalStateException("Queue full");            enqueue(new Node
   
    (e));            ++n;        }        count.set(n);    } finally {
           putLock.unlock();    }}
   

4 新增

以 offer 方法为例.

public boolean offer(E e) {
      if (e == null) throw new NullPointerException();   // 如果“队列已满”，则返回false，表示插入失败。   final AtomicInteger count = this.count;  if (count.get() == capacity) return false;   int c = -1;   // 新建“节点e”   Node
   
     node = new Node(e);   final ReentrantLock putLock = this.putLock;   // 获取“插入锁putLock”   putLock.lock();   try {
        // 再次对“队列是不是满”的进行判断。     // 若“队列未满”，则插入节点。     if (count.get() < capacity) {
          // 插入节点 enqueue(node);      // 将“当前节点数量”+1，并返回“原始的数量”       c = count.getAndIncrement();       // 如果在插入元素之后，队列仍然未满，则唤醒notFull上的等待线程。       if (c + 1 < capacity)  notFull.signal();     }   } finally {
          // 释放“插入锁putLock”       putLock.unlock();  }   // 如果在插入节点前，队列为空；则插入节点后，唤醒notEmpty上的等待线程   if (c == 0)     signalNotEmpty();     return c >= 0;  }
   

enqueue

• 将node添加到队列末尾，并设置node为新的尾节点
  

signalNotEmpty

• 发出等待信号。 仅能从put/offer调用（否则通常不锁定takeLock）,唤醒notEmpty上的等待线程.
  

新增数据成功后，在适当时机，会唤起 put 的等待线程（队列不满时），或者 take 的等待线程（队列不为空时），这样保证队列一旦满足 put 或者 take 条件时，立马就能唤起阻塞线程，继续运行，保证了唤起的时机不被浪费。

5 取出

以take()为例.取出并返回队列的头。若队列为空，则一直等待。

public E take() throws InterruptedException {
      E x;   int c = -1;   final AtomicInteger count = this.count;   final ReentrantLock takeLock = this.takeLock;   // 获取“取出锁”，若当前线程是中断状态，则抛出InterruptedException异常   takeLock.lockInterruptibly();   try {
        // 若“队列为空”，则一直等待。     while (count.get() == 0) {
          notEmpty.await();     }     // 取出元素     x = dequeue();     // 取出元素之后，将“节点数量”-1；并返回“原始的节点数量”。     c = count.getAndDecrement();    if (c > 1) notEmpty.signal();  } finally {
        // 释放“取出锁”     takeLock.unlock();   }   // 如果在“取出元素之前”，队列是满的；则在取出元素之后，唤醒notFull上的等待线程。   if (c == capacity)     signalNotFull();   return x;}

dequeue

• 删除队列的头节点，并将表头指向“原头节点的下一个节点”。
  

signalNotFull

• 发出等待的信号。 仅能从take/poll调用.唤醒notFull上的等待线程.
  

6 总结

理解好阻塞队列很重要,经常会被用在线程池的场景中.

转载地址：https://javaedge.blog.csdn.net/article/details/105624698 如侵犯您的版权，请留言回复原文章的地址，我们会给您删除此文章，给您带来不便请您谅解！