BlockingQueue 的使用与源码阅读

柔情只为你懂 2022-12-23 09:43 180阅读 0赞

### 文章目录 ###

*   *  1.BlockingQueue
     *  2.BlockingQueue 出队入队使用方法
     *  3.使用例子
     *  4.ArrayBlockingQueue结构
     *   *  ArrayBlockingQueue的使用
         *  ArrayBlockingQueue继承结构
         *  例子中流程示意图
         *  4.4 核心属性
         *  4.5 构造方法
         *  4.6 void put(E e)
         *  4.7E take()
     *  5 LinkedBlockingQueue结构
     *   *  5.1 核心属性
         *  5.2 构造方法
         *  5.3 void put(E e)
         *  5.4 E take()
     *  6.当生产者线程速度大于消费线程，造成队列满的情况
     *  7 当生产者线程速度小于消费线程，造成队列空的情况
     *  8 总结

## 1.BlockingQueue ##

*  ArrayBlockingQueue  
    使用数组结构实现的阻塞队列
 *  LinkedBlockingQueue  
    使用链表结构实现的阻塞队列

## 2.BlockingQueue 出队入队使用方法 ##

<table> 
 <thead> 
  <tr> 
   <th>现象</th> 
   <th>阻塞</th> 
   <th>返回Null</th> 
   <th>超时</th> 
   <th>抛异常</th> 
  </tr> 
 </thead> 
 <tbody> 
  <tr> 
   <td>入队</td> 
   <td>Put(o)</td> 
   <td>Offer(o)</td> 
   <td>offer(o,timeout,timeunit)</td> 
   <td>add(o)</td> 
  </tr> 
  <tr> 
   <td>出队</td> 
   <td>Take()</td> 
   <td>Poll()</td> 
   <td>poll(timeout, timeunit)</td> 
   <td>remove()</td> 
  </tr> 
  <tr> 
   <td>访问对头</td> 
   <td></td> 
   <td>Peek()</td> 
   <td></td> 
   <td>element()</td> 
  </tr> 
 </tbody> 
</table>

为什么是这种情况。先看其继承结构。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70_pic_center]从下至上 ArrayBlockingQueue 继承了BlockingQueue，Queue，Collection。有没有发现上表的方法，**阻塞的入队出队方法是BlockingQueue接口的**，返回**Null的接口是Queue**，**抛异常的方法是Collection接口的**（element除外，其内部调用peek方法，为Null手动抛异常），这些在接口设计上就已经设定好的。

## 3.使用例子 ##

ArrayBlockingQueue的构造方法里没有无参构造函数，因为他是有界的需要你指定其大小容量。

public static void main(String[] args) { 
       BlockingQueue<Integer> abq = new ArrayBlockingQueue<Integer>(3);
       new Thread(new Runnable() { 
           @Override
           public void run() { 
               for (int i = 0; i < 6; i++) { 
                   try { 
                       abq.put(i);
                       System.out.println("put:" + i);
                   } catch (InterruptedException e) { 
                       e.printStackTrace();
                   }
               }
           }
       }).start();
       new Thread(new Runnable() { 
           @Override
           public void run() { 
               try { 
                    Thread.sleep(3000);
                    System.out.println("sleep3000");
                } catch (InterruptedException e) { 
                    e.printStackTrace();
               }
               for (; ; ) { 
                   try { 
                       System.out.println("take:" + abq.take());
                   } catch (InterruptedException e) { 
                       e.printStackTrace();
                   }
               }
           }
       }).start();
    }

控制台输出

> put:0  
> put:1  
> put:2  
> sleep3000  
> take:0  
> take:1  
> put:3  
> take:2  
> put:4  
> take:3  
> put:5  
> take:4  
> take:5

可以很明显的看到当生产者线程在存放第四个元素的时**没存放进去**而是一直**等待消费者线程消费了其中的元素后**，其内部元素量小于队列容量才再次put进去。  
生产者线程在存放第四个元素时，先获取锁，判断已存量==队列长度，**调用notFull.await()方法使生产线程进入等待状态这个过程会释放锁**，**当消费线程消费后调用notFull.signal();会唤醒正在等待状态的生产线程**。生产线程在获取到锁再存放元素。

## 4.ArrayBlockingQueue结构 ##

其中数组是循环使用的，真实中不存在从3到1的指针，使用的是putIndex，takeIndex两个下标索引。当此时存放的元素putIndex等于数组的长度时，putIndex下次就会变成0，又从头开始存放元素。takeIndex原理一样。  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70_pic_center 1]  
基于数组实现的，并且内部持有一把ReentrantLock（读写一把）。

### ArrayBlockingQueue的使用 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70_pic_center 2]  
简单说一下情况，后面从方法中分析。生产线程将元素放入队列中时需要先获取锁，获取到锁后向队列存放元素。消费线程获取队列中元素时也需要先获取锁，获取到锁后获取队列中的元素。

### ArrayBlockingQueue继承结构 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70_pic_center 3]

### 例子中流程示意图 ###

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70_pic_center 4]

### 4.4 核心属性 ###

*  **final Object\[\] items;**  
    用来存放数据元素的数组，结构图中红色部分
 *  **int takeIndex;**  
    下次take出队时，元素的下标索引
 *  **int putIndex;**  
    跟takeIndex对应，下次入队是的下标索引
 *  **int count;**  
    队列中的元素个数
 *  **final ReentrantLock lock;**  
    结构图中蓝色的可重入锁
 *  private final Condition notEmpty;
 *  private final Condition notFull;  
    Condition接口也提供了类似Object的监视器方法，与Lock配合可以实现等待/通知模式。当队满或者队空通知等待用

### 4.5 构造方法 ###

public ArrayBlockingQueue(int capacity, boolean fair) { 
        if (capacity <= 0)
            throw new IllegalArgumentException();
         //初始化数组容量
        this.items = new Object[capacity];
        //指定锁的类型为公平锁还是非公平锁
        lock = new ReentrantLock(fair);
        notEmpty = lock.newCondition();
        notFull =  lock.newCondition();
    }

//默认为非公平锁
    public ArrayBlockingQueue(int capacity) { 
        this(capacity, false);
    }

### 4.6 void put(E e) ###

public void put(E e) throws InterruptedException { 
            //检查一下插入节点非空
            checkNotNull(e);
            final ReentrantLock lock = this.lock;
            //加锁
            lock.lockInterruptibly();
            try { 
                //如果队列满了线程进入等待状态释放锁
                while (count == items.length)
                    notFull.await();
                //入队操作
                enqueue(e);
            } finally { 
                //释放锁
                lock.unlock();
            }
        }

enqueue是入队操作，一定是在已经获得了锁的情况下再进行操作的，并且判断对内元素小于队列容量

private void enqueue(E x) { 
        final Object[] items = this.items;
        //元素存入数组
        items[putIndex] = x;
        //在结构处说明了。因为数组是循环使用的，那么下次存放的位置就在0
        if (++putIndex == items.length)
            putIndex = 0;
        count++;
        //有元素入队了，通知消费线程消费
        notEmpty.signal();
    }

### 4.7E take() ###

public E take() throws InterruptedException { 
        final ReentrantLock lock = this.lock;
        lock.lockInterruptibly();
        try { 
            //队列中没有元素，线程进入等待状态释放锁
            while (count == 0)
                notEmpty.await();
            return dequeue();
        } finally { 
            lock.unlock();
        }
    }

dequeue是出队操作，一定是在已经获得了锁的情况下再进行操作的，并且判断对内元素大于0

private E dequeue() { 
            final Object[] items = this.items;
            //取得元素
            E x = (E) items[takeIndex];
            //置为空
            items[takeIndex] = null;
            //在结构处说明了。因为数组是循环使用的，那么队头在0的位置
            if (++takeIndex == items.length)
                takeIndex = 0;
            count--;
            //迭代器的一些处理
            if (itrs != null)
                itrs.elementDequeued();
            //有元素被消费了，通知生产线程生产
            notFull.signal();
            return x;
        }

## 5 LinkedBlockingQueue结构 ##

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70_pic_center 5]

### 5.1 核心属性 ###

**- 节点结构**

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

**- private final int capacity;**  
容量  
**- private final AtomicInteger count = new AtomicInteger();**  
已存在的个数  
**- transient Node head;**  
头节点  
**- private transient Node last;**  
尾节点  
**- private final ReentrantLock takeLock = new ReentrantLock();**  
入队锁  
**- private final Condition notEmpty = takeLock.newCondition();**  
队伍不空  
**- private final ReentrantLock putLock = new ReentrantLock();**  
出队锁  
**- private final Condition notFull = putLock.newCondition();**  
队伍不满

### 5.2 构造方法 ###

有给定容量按参数作为容量，要不以Integer.MAX\_VALUE作为容量

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

![初始化时队列元素][20201125180710592.png_pic_center]

1.  从上面的属性大概能推断每个添加到LinkedBlockingQueue队列中的数据都活构造成一个Node节点，添加到链表的尾部。
2.  其中head和last分别指向队列的头结点和尾结点。
3.  与ArrayBlockingQueue不同的是，LinkedBlockingQueue内部分别使用了takeLock 和 putLock添加和删除操作并不是互斥操作，可以同时进行。
4.  **这里如果不指定队列的容量大小，也就是使用默认的Integer.MAX\_VALUE，如果存在添加速度大于删除速度时候，有可能会内存溢出，这点在使用前希望慎重考虑**。
5.  另外，LinkedBlockingQueue对每一个lock锁都提供了一个Condition用来挂起和唤醒其他线程。

### 5.3 void put(E e) ###

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 { 
                //如果已存在个数等于容量线程进入等待状态，等待notFull.signal();唤醒
                while (count.get() == capacity) { 
                    notFull.await();
                }
                //入队，操作很简单，尾指针的next指向新构造的节点，使尾指针指向新构造的节点
                //last = last.next = node;
                enqueue(node);
                //返回当前数量。并且原子自增1
                c = count.getAndIncrement();
                //已存在元素小于容量，继续唤醒线程put元素
                if (c + 1 < capacity)
                    notFull.signal();
            } finally { 
                putLock.unlock();
            }
            //刚刚put进去一个元素，唤醒take线程进行消费
            if (c == 0)
                signalNotEmpty();
        }

enqueue方法  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70_pic_center 6]

### 5.4 E take() ###

public E take() throws InterruptedException { 
            E x;
            int c = -1;
            final AtomicInteger count = this.count;
            final ReentrantLock takeLock = this.takeLock;
            takeLock.lockInterruptibly();
            try { 
                //个数为空，线程进入等待状态，等待notFull.signal();唤醒
                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;
        }

/** *从队列头删除节点 */
    private E dequeue() { 
        // 获取到head节点
        Node<E> h = head;
        //头节点:head节点指向的下一个节点
        Node<E> first = h.next;
        //head节点next指向自己，等待下次gc回收
        h.next = h;
        // head节点指向新的节点
        head = first;
        // 获取到新的head节点的item值
        E x = first.item;
        // 新head节点的item值设置为null
        first.item = null;
        return x;
    }

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70_pic_center 7]

## 6.当生产者线程速度大于消费线程，造成队列满的情况 ##

生产者线程ABC依次获取锁，当队列满时（count.get() == capacity）线程进入等待状态，等待唤醒。在消费线程E进行消费时，判断队列还有元素继续唤醒消费线程进行消费。**当线程判断在自己消费之前队列为满，会在消费完之后唤醒位于等待池中的生产线程中的一个进行生产。**  
![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70_pic_center 8]

## 7 当生产者线程速度小于消费线程，造成队列空的情况 ##

嗯，跟上面查不多只是相反。消费者消费数据时队列为空都调用notEmpty.await();线程进入等待状态等待唤醒。在生产者线程进行生产时，判断队列还不满，唤醒生产者线程继续进行生产。当某个生产者线程判断在自己生产之前队列为空，会在生产之后唤醒位于等待池中的消费线程中的一个进行消费。

## 8 总结 ##

1.  是否有界无界不同  
    ArrayBlockingQueue是有界的**初始化必须指定大小**，而LinkedBlockingQueue可以是有界的也可以是无界的(Integer.MAX\_VALUE),**无界时需要注意内存溢出问题**。
2.  锁的使用不同  
    ArrayBlockingQueue内部由一个ReentrantLock来实现出入队列的线程安全。LinkedBlockingQueue内部由两个ReentrantLock来实现出入队列的线程安全，由各自的Condition对象来实现通知等待功能，使用两把锁能提高队列的吞吐量，生产者和消费者可以并行地操作队列中的数据。（遗留问题：为什么ArrayBlockingQueue使用一把锁）
3.  数据存储的结构不同  
    ArrayBlockingQueue采用的是数组作为数据存储容器，而LinkedBlockingQueue采用的则是以Node节点的链表。数组的存储容器，在插入或删除元素时不会产生或销毁任何额外的对象实例，而LinkedBlockingQueue则会生成一个额外的Node对象。

[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70_pic_center]: /images/20221120/52a5d4f5fcaa46249e55e9f239b99322.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70_pic_center 1]: /images/20221120/9050e09dde234479b9e60baee01794d3.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70_pic_center 2]: /images/20221120/d2cbfdd934da461e8251838036224f7f.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70_pic_center 3]: /images/20221120/558906060597454fa1394daf4a1223c0.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70_pic_center 4]: /images/20221120/acc23e8f13bc438a82fc8298c6bd40e1.png
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70_pic_center 5]: https://img-blog.csdnimg.cn/20201125162248529.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ==,size_16,color_FFFFFF,t_70#pic_center
[20201125180710592.png_pic_center]: https://img-blog.csdnimg.cn/20201125180710592.png#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70_pic_center 6]: https://img-blog.csdnimg.cn/20201125182437909.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70_pic_center 7]: https://img-blog.csdnimg.cn/20201125182913117.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ==,size_16,color_FFFFFF,t_70#pic_center
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ_size_16_color_FFFFFF_t_70_pic_center 8]: https://img-blog.csdnimg.cn/20201125165759299.png?x-oss-process=image/watermark,type_ZmFuZ3poZW5naGVpdGk,shadow_10,text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dzZGZ5bQ==,size_16,color_FFFFFF,t_70#pic_center