Java多线程-BlockingQueue

Posted 2021-03-14 reupe

　　

• BlockingQueue的继承结构

　　BlockingQueue是线程安全的阻塞队列，当队列为空时，拉取队列的线程会等待队列中重新有元素；当队列满时，添加元素的线程会等待队列有空位储存新元素。BlockingQueue的继承接口如下:

 

• 生产者-消费者模式

　　ArrayBlokingQueue实现类需要设置固定的大小，SynchronousQueue只有一个容量，而LinkedBlockingQueue是可变容量的队列。一般而言，BlockingQueue应用于producer-consumer场景，即生产者-消费者，顾名思义，生产者就是向队列中添加元素的线程，消费者就是从队列中取出元素的线程，代码写法一般模板如下:

/**
 * 生产者
 */  
class Producer implements Runnable {
    private final BlockingQueue queue;
    Producer(BlockingQueue q) { queue = q; }
    public void run() {
      try {
        while (true) { queue.put(produce()); }
      } catch (InterruptedException ex) { ... handle ...}
    }
    Object produce() { ... }
 }
 
/*
 * 消费者
 */
 class Consumer implements Runnable {
    private final BlockingQueue queue;
    Consumer(BlockingQueue q) { queue = q; }
    public void run() {
      try {
        while (true) { consume(queue.take()); }
      } catch (InterruptedException ex) { ... handle ...}
    }
    void consume(Object x) { ... }
 }
 
  class Setup {
    void main() {
      BlockingQueue q = new SomeQueueImplementation();
      Producer p = new Producer(q);
      Consumer c1 = new Consumer(q);
      Consumer c2 = new Consumer(q);
      new Thread(p).start();
      new Thread(c1).start();
      new Thread(c2).start();
    }

 

需要注意的就是生产者和消费者一定要作用于同一个阻塞队列。

• BlockingQueue存储数据的数据结

　　LinkedBlockingQueue的内部存储结构是链表，定义了一个内嵌类，

 1 /**
 2      * Linked list node class
 3      */
 4     static class Node<E> {
 5         E item;
 6 
 7         /**
 8          * One of:
 9          * - the real successor Node
10          * - this Node, meaning the successor is head.next
11          * - null, meaning there is no successor (this is the last node)
12          */
13         Node<E> next;
14 
15         Node(E x) { item = x; }
16     }

可以看出这是一个单向链表，因为每个Node节点只保存有当前节点的值，以及指向下一个Node节点的引用。类似地，ArrayBlockingQueue根据名称，可以推断出，其内部储存结构是数组，这里不再赘述。

　　当构造LinkedBlockingQueue时，采用默认的构造器，将会创建一个最大节点数为Integer.MAX_VALUE的队列，并且会创建一个Node节点对象，其值为null，last和head引用均指向之，这就完成了

 1 /**
 2    * Creates a {@code LinkedBlockingQueue} with a capacity of
 3    * {@link Integer#MAX_VALUE}.
 4    */
 5 public LinkedBlockingQueue() {
 6      this(Integer.MAX_VALUE);
 7 }
 8 
 9 
10 public LinkedBlockingQueue(int capacity) {
11      if (capacity <= 0) throw new IllegalArgumentException();
12      this.capacity = capacity;
13      last = head = new Node<E>(null);
14 }

队列的初始化。之后就可以put和take了。

 

• put和take时的线程安全实现

 　　首先看put()方法，即向队列中放入元素，是怎样实现线程安全的。

 public void put(E e) throws InterruptedException {
        if (e == null) throw new NullPointerException();
        int c = -1;
        // 封装元素为一个新的节点
        Node<E> node = new Node<E>(e);
        // 使用put重入锁对象
        final ReentrantLock putLock = this.putLock;
        final AtomicInteger count = this.count;
        // 可以被中断的blocking
        putLock.lockInterruptibly();
        try {
            // point1: 如果当前元素数量达到队列最大容量，就释放锁并挂起当前线程，直到被 notFull.signal()唤醒
            while (count.get() == capacity) {
                notFull.await();
            }
            // 节点入队
            enqueue(node);
            // 入队前队列的数量+1如果小于容器最大容量，就调用 notFull.signal()唤醒上面代码中wait的线程
            c = count.getAndIncrement();
            if (c + 1 < capacity)
                notFull.signal();
        } finally {
            putLock.unlock();   // 释放锁
        }
        // 这段代码是为了唤醒take()中挂起的线程，具体原因下面详解
        if (c == 0)
            signalNotEmpty();
    }

put方法保证线程安全是基于重入锁机制，比较容易理解，假设线程A执行到point1, 如果当前链表容量达到最大，那么就进入while中挂起线程，否则就继续进行下去。

假设有这么一个场景，线程A执行put，此时队列是满的，那么线程A就会在point1处挂起，那么谁来唤醒线程A？ 答案是take()方法，看下面take方法

 1 public E take() throws InterruptedException {
 2         E x;
 3         int c = -1;
 4         final AtomicInteger count = this.count;
 5         final ReentrantLock takeLock = this.takeLock;
 6         takeLock.lockInterruptibly();
 7         try {
 8             // piont1: 如果当前队列为空，则挂起线程并释放锁
 9             while (count.get() == 0) {
10                 notEmpty.await();
11             }
12             // 末尾元素出队
13             x = dequeue();
14             // 如果队列容量不为空，则唤醒处于等待中的take线程
15             c = count.getAndDecrement();
16             if (c > 1)
17                 notEmpty.signal();
18         } finally {
19             takeLock.unlock();
20         }
21         // point2: c为take前的容量，即当前容量为c-1, 唤醒等待中的put线程
22         if (c == capacity)
23             signalNotFull();
24         return x;
25     }
26 
27 
28     private void signalNotFull() {
29         final ReentrantLock putLock = this.putLock;
30         putLock.lock();
31         try {
32             notFull.signal();
33         } finally {
34             putLock.unlock();
35         }
36     }

前面假设了当前队列是满的，那么put线程A已经阻塞在了put()方法中的point1位置，直到线程B执行了take()方法，取走一个元素，然后执行signalNotFull方法，唤醒put线程。   而put()方法中的signalNotEmpty()方法刚好相反，是在容器为0时，有线程先执行了take()阻塞，直到put去唤醒take线程。

 

　　从上面可以看出来，BlockingQueue使用重入锁来保证线程安全，使用Condition对象的await()和sigal()协调线程之间的合作以达到线程安全的阻塞队列的效果，AtomicInteger对象count是put和take之间的重要桥梁，它代表了当前队列元素个数，保证获取增加元素个数的原子性，没有它就无从保证数据的正确。实现中还有很多细节，代码中都考虑进去了，比如当容器为空的时候，容器满的时候，容器即不为空也不为满的时候，那么signalNotFull()和signalNotEmpty压根就不会执行了。