LinkedBlockingQueue和ArrayBlockingQueue

深藏阁楼爱情的钟 2022-09-03 01:17 113阅读 0赞

#### BlockingQueue接口 ####

BlockingQueue是一个先进先出带阻塞功能的队列，当入队列时，若队列已满，则阻塞调用者;当出队列时，若队列为空，则阻塞调用者。

public interface BlockingQueue<E> extends Queue<E> {
        
        //...
        boolean add(E e);
        boolean offer(E e);
        void put(E e) throws InterruptedException;
        boolean remove(Object o);
        E take() throws InterruptedException;
        E poll(long timeout, TimeUnit unit) throws InterruptedException;
        //...
    }

入队列的方法中，add()和offer()是无阻塞的，返回的是布尔类型，也是Queue的接口，而put()是阻塞的,并且可以被中断。当队列为满的时候，add()会抛出异常，offer()则直接返回false，put()会阻塞。

出队列的方法中，remove是非阻塞的，take和poll是阻塞的。

#### ArrayBlockingQueue类 ####

ArrayBlockingQueue是一个用数组实现的环形队列，在构造方法中，会要求传入数组的容量。向这个队列中添加元素的下标如果超过了数组的size-1，那么这个元素会重新从下标\[0\]开始添加

ArrayBlockingQueue的构造方法

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();
    }
    /**
    * capacity：容量
    * fair：是否是一个公平的队列
    * c：初始化的时候会把c中的元素添加到队列中去
    */
    public ArrayBlockingQueue(int capacity, boolean fair, Collection<? extends E> c) {
        
       	this(capacity, fair);
    	// ...
    }

核心数据结构

public class ArrayBlockingQueue<E> extends AbstractQueue<E> implements BlockingQueue<E>, java.io.Serializable {
        
    	//...
    	//存放元素的数组
    	final Object[] items; 
    	// 队头指针，生产者用
    	int takeIndex;
    	// 队尾指针，消费者用
    	int putIndex;
    	//队列中有多少个元素
    	int count;
    	// 核心为1个锁两个条件
    	final ReentrantLock lock; 
    	private final Condition notEmpty; 
    	private final Condition notFull; 
    	//...
    }

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();
        }
    }
    
    private void enqueue(E x) {
        
        final Object[] items = this.items;
        items[putIndex] = x;
        //ArrayBlockingQueue是一个环形队列，当putIndex+1达到数组长度长度时，从下标0开始put
        if (++putIndex == items.length)
            putIndex = 0;
        count++;
        //元素插入队列后通知notEmpty，即通知消费者消费
        notEmpty.signal();
    }

take方法

public E take() throws InterruptedException {
        
        final ReentrantLock lock = this.lock;
        //获得一个可中断的锁
        lock.lockInterruptibly();
        try {
        
            while (count == 0)
            	//如果队列空了，notEmpty条件阻塞
                notEmpty.await();
            return dequeue();
        } finally {
        
            lock.unlock();
        }
    }
    
    private E dequeue() {
        
        final Object[] items = this.items;
        @SuppressWarnings("unchecked")
        E x = (E) items[takeIndex];
        //取出了元素以后设置为null
        items[takeIndex] = null;
        //同put一样，如果takeIndex的下标+1和数组的容量相对，那么重新从下标[0]开始消费
        if (++takeIndex == items.length)
            takeIndex = 0;
        count--;
        if (itrs != null)
            itrs.elementDequeued();
        //消费了队列以后通知notFull，即通知生产者生产
        notFull.signal();
        return x;
    }

#### LinkedBlockingQueue ####

LinkedBlockingQueue是一种基于单向链表的阻塞队列。因为队头和队尾是2个指针分开操作的， 所以用了2把锁+2个条件，同时有1个AtomicInteger的原子变量记录count数。（ArrayBlockingQueue中take和put操作的同一个数组所以可以用一把锁，链表中take和put分别操作的是队列头和队列尾两个元素，所以用了两把锁）

public class LinkedBlockingQueue<E> extends AbstractQueue<E> implements BlockingQueue<E>, java.io.Serializable {
        
    	// ...
    	//容量
    	private final int capacity;
    	// 原子变量 记录有多少个元素
    	private final AtomicInteger count = new AtomicInteger(0); 
    	// 单向链表的头部
    	private 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(); // ...
    }

capacity默认最大值

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

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;
        //自己put的时候不让别人put
        putLock.lockInterruptibly();
        try {
        
            while (count.get() == capacity) {
        
            	//队列满的时候阻塞，await的时候会释放持有的锁
                notFull.await();
            }
            //把元素放进队列
            enqueue(node);
            //获得count，然后count的值+1
            c = count.getAndIncrement();
            if (c + 1 < capacity)
            	//如果队列还没满，则通知其他put线程
                notFull.signal();
        } finally {
        
            putLock.unlock();
        }
        if (c == 0)
        	//这里的c是count+1之前获得的，这里的count其实已经非空了
        	//通知消费者消费
            signalNotEmpty();
    }
    
    private void signalNotEmpty() {
        
        final ReentrantLock takeLock = this.takeLock;
        takeLock.lock();
        try {
        
        	//任务在可以调用await(),signal(),signalAll()之前，必须拥有这个锁。
            notEmpty.signal();
        } finally {
        
            takeLock.unlock();
        }
    }

take方法

public E take() throws InterruptedException {
        
         E x;
         int c = -1;
         final AtomicInteger count = this.count;
         final ReentrantLock takeLock = this.takeLock;
         //获取take锁，自己take的时候不让别人take
         takeLock.lockInterruptibly();
         try {
        
             while (count.get() == 0) {
        
             	//await的时候会释放持有的锁
                 notEmpty.await();
             }
             //获取元素
             x = dequeue();
             //获取count后，将count-1
             c = count.getAndDecrement();
             if (c > 1)
             	 //如果还有元素，通知其他take线程
                 notEmpty.signal();
         } finally {
        
             takeLock.unlock();
         }
         //这里c是count-1前获得的，所以这里的count已经-1了，即队列没有满
         if (c == capacity)
             //如果队列没有满，通知生产者生产
             signalNotFull();
         return x;
    }
    
    private void signalNotFull() {
        
        final ReentrantLock putLock = this.putLock;
        putLock.lock();
        try {
        
        	//任务在可以调用await(),signal(),signalAll()之前，必须拥有这个锁。
            notFull.signal();
        } finally {
        
            putLock.unlock();
        }
    }