Java并发 之 阻塞队列LinkedBlockingQueue与ArrayBlockingQueue

常用的普通队列: ArrayList 和 LinkedList，都实现了List接口，实现的方式不同。ArrayList基于数组实现，查找时效率更高，LinkedList基于链表结构实现，插入和删除效率更高。

阻塞队列相比于 普通队列最大的不同点：阻塞队列支持阻塞添加和阻塞删除。

• 阻塞添加是指当阻塞队列元素已满时，队列会阻塞加入元素的线程，直队列元素不满时才重新唤醒线程执行元素加入操作。
• 阻塞删除是指在队列元素为空时，删除队列元素的线程将被阻塞，直到队列不为空再执行删除操作(可返回被删除的元素)。

阻塞队列接口 BlockingQueue 继承自Queue接口：

public interface BlockingQueue<E> extends Queue<E> {

	// 将指定的元素插入到此队列的尾部
	// 插入成功返回 true，如果此队列已满，则抛IllegalStateException异常 
	boolean add(E e); 
	
	// 将指定的元素插入到队列的尾部 
	// 如果该队列已满，则在到达指定的等待时间之前等待可用的空间,该方法可中断 
	boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException; 
	
	//将指定的元素插入到队列的尾部，如果该队列已满，则一直等待（阻塞）。 
	void put(E e) throws InterruptedException; 
	
	// 获取并移除此队列的头部，如果没有元素则等待（阻塞），直到有元素将唤醒等待线程执行该操作 
	E take() throws InterruptedException; 
	
	// 获取并移除此队列的头部，在指定的等待时间前一直等到获取元素，超过时间方法将结束
    // 可中断等待操作
	E poll(long timeout, TimeUnit unit) throws InterruptedException; 
	
	//从此队列中移除指定元素的单个实例（如果存在），不存在则抛出异常
	boolean remove(Object o); 
}

	//除了上述方法还有继承自Queue接口的方法 
	//获取但不移除此队列的头,没有则跑异常NoSuchElementException 
	E element(); 
	
	//获取但不移除此队列的头；如果此队列为空，则返回 null。 
	E peek(); 
	
	//获取并移除此队列的头，如果此队列为空，则返回 null。 
	E poll();

通常情况下我们都是通过上述方法对阻塞队列的元素进行增删查操作。

ArrayBlockingQuere的基本使用和源码分析：

通过ArrayBlockingQueue队列实现一个生产者消费者的例子：

consumer消费者和producer生产者，consumer通过 take() 获取头元素，队列为空则一直阻塞等待，producer通过 put() 向队列尾部添加元素，若队列已满，则一直阻塞等待。

public class BlockQueueDemo {
    private static ArrayBlockingQueue<Goods> queue = new ArrayBlockingQueue<>(3);

    public static void main(String[] args) {
            new Thread(new Producer(queue)).start();
            new Thread(new Consumer(queue)).start();
            new Thread(new Producer(queue)).start();
            new Thread(new Consumer(queue)).start();
    }
}

class Goods{
    public Goods() {

    }
}
class Producer implements Runnable{
    private ArrayBlockingQueue<Goods> queue;

    public Producer(ArrayBlockingQueue<Goods> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        while (true) {
            produce();
        }
    }
    void produce(){
        try {
            Goods goods = new Goods();
            queue.put(goods);
            System.out.println("生产商品");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }

    }
}
class Consumer implements Runnable{
    private ArrayBlockingQueue<Goods> queue;

    public Consumer(ArrayBlockingQueue<Goods> queue) {
        this.queue = queue;
    }

    @Override
    public void run() {
        while (true){
            consume();
        }
    }
    void consume(){
        try {
            TimeUnit.MILLISECONDS.sleep(1000);
            queue.take();
            System.out.println("消费商品");
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }
}

ArrayBlockingQueue内部的阻塞队列是通过重入锁ReentrantLock和Condition条件队列实现的，所以ArrayBlockingQueue中的元素存在公平访问与非公平访问的区别。创建 公平和非公平阻塞队列：

//默认非公平阻塞队列
ArrayBlockingQueue queue = new ArrayBlockingQueue(2);
//公平阻塞队列
ArrayBlockingQueue queue1 = new ArrayBlockingQueue(2,true);

//构造方法源码
public ArrayBlockingQueue(int capacity) {
     this(capacity, false);
 }

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

ArrayBlockingQueue 的内部是通过一个可重入锁ReentrantLock和两个Condition条件对象来实现阻塞。

 public class ArrayBlockingQueue<E> extends AbstractQueue<E>
        implements BlockingQueue<E>, java.io.Serializable {

    // 存储数据的数组 
    final Object[] items;

    // 获取数据的索引，指向头元素 
    int takeIndex;

    // 添加数据的索引，指向尾元素
    int putIndex;

    // 队列元素的个数 
    int count;


    // 控制并发访问的锁 
    final ReentrantLock lock;

    // 条件队列对象，用于通知take方法队列已有元素，可执行获取操作 
    private final Condition notEmpty;

    // 条件队列对象，用于通知put方法队列未满，可执行添加操作 
    private final Condition notFull;

    // 迭代器
    transient Itrs itrs = null;

}

put()，阻塞添加：如果队列元素已满，那么当前线程将会被 notFull 条件对象挂起加到等待队列中，直到队列有空档才会唤醒执行添加操作。但如果队列没有满，那么就直接调用enqueue(e)方法将元素加入到数组队列中。

入队操作：

private void enqueue(E x) {
    //获取当前数组
    final Object[] items = this.items;
    //通过putIndex索引对数组进行赋值
    items[putIndex] = x;
    //索引自增，如果已是最后一个位置，重新设置 putIndex = 0;
    if (++putIndex == items.length)
        putIndex = 0;
    count++;//队列中元素数量加1
    // 唤醒调用take()方法的线程，执行元素获取操作。
    notEmpty.signal();
}

//put方法，阻塞时可中断
 public void put(E e) throws InterruptedException {
     checkNotNull(e);
      final ReentrantLock lock = this.lock;
      lock.lockInterruptibly();
      try {
          //当队列元素个数与数组长度相等时，无法添加元素
          while (count == items.length)
              //将当前调用线程挂起，添加到notFull条件队列中等待唤醒
              notFull.await();
          enqueue(e);
      } finally {
          lock.unlock();
      }
  }

take()，阻塞移除，如果队列没有数据那么就加入notEmpty条件队列等待(有数据就直接取走，方法结束)，如果有新的put线程添加了数据，那么put操作将会唤醒take线程，执行take操作。

出队操作：

//删除队列头元素并返回
 private E dequeue() {
     // 拿到当前数组的数据
     final Object[] items = this.items;
      @SuppressWarnings("unchecked")
      // 获取要删除的对象
      E x = (E) items[takeIndex];
      // 将数组中takeIndex索引位置设置为null
      items[takeIndex] = null;
      //takeIndex索引加1并判断是否与数组长度相等，
      // 如果相等说明已到尽头，恢复为0
      if (++takeIndex == items.length)
          takeIndex = 0;
      count--;//队列个数减1
      if (itrs != null)
          itrs.elementDequeued();
      //删除了元素说明队列有空位，唤醒notFull条件对象添加线程，执行添加操作
      notFull.signal();
      return x;
    }

//从队列头部删除，队列没有元素就阻塞，可中断
 public E take() throws InterruptedException {
    final ReentrantLock lock = this.lock;
      lock.lockInterruptibly();//中断
      try {
          //如果队列没有元素
          while (count == 0)
              //执行阻塞操作
              notEmpty.await();
          return dequeue();
      } finally {
          lock.unlock();
      }
    }

LinkedBlockingQuere的源码分析：

LinkedBlockingQueue 内部成员：

public class LinkedBlockingQueue<E> extends AbstractQueue<E>
        implements BlockingQueue<E>, java.io.Serializable {

    /**
     * 节点类，用于存储数据
     */
    static class Node<E> {
        E item;

        /**
         * One of:
         * - the real successor Node
         * - this Node, meaning the successor is head.next
         * - null, meaning there is no successor (this is the last node)
         */
        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();

    // 条件对象，当队列没有数据时用于挂起执行删除的线程 
    private final Condition notEmpty = takeLock.newCondition();

    // 添加元素时使用的锁
    private final ReentrantLock putLock = new ReentrantLock();

    // 条件对象，当队列数据已满时用于挂起执行添加的线程 
    private final Condition notFull = putLock.newCondition();

}

每个添加到LinkedBlockingQueue队列中的数据都将被封装成Node节点，添加的链表队列中，其中head和last分别指向队列的头结点和尾结点。

与ArrayBlockingQueue不同的是，LinkedBlockingQueue内部分别使用了 takeLock 和 putLock 对并发进行控制，也就是说，添加和删除操作并不是互斥操作，可以同时进行，这样也就可以大大提高吞吐量。
 

 

 

 

本文地址：https://blog.csdn.net/wcfcangzhuo/article/details/107271198