ArrayBlockingQueue源码解读

柔情只为你懂 2021-09-02 09:06 462阅读 0赞

#### ArrayBlockingQueue ####

它是BlockingQueue阻塞队列基于数组的一种实现。  
**阻塞队列**：首先它是一个队列queue，同时如果队列为空，从队列取出元素的操作会被阻塞；如果队列为满，向队列中增加元素的操作会被阻塞。  
**Queue的常见api辨析：**

*  boolean add(E e); // 向队列中增加元素，必须要加入进去。如果队列已满，抛出IllegalStateException异常
 *  boolean offer(E e);//本意是提供，意思是你要就拿走，不要告诉我一声。也是向队列增加元素，但如果队列满，返回false，不是抛出异常
 *  E remove();// 移除队列中第一个元素，必须要移除，如果队列为空，抛出NoSuchElementException
 *  E poll();// 本意是轮询，没有remove那么强烈。它是移除第一个元素，如果队列为空，返回null；
 *  E peek();// 本意是窥视，只是看看第一个元素，不操作。返回第一个元素。

**BlockingQueue中常见api辨析：**

*  void put(E e) throws InterruptedException;//本意是将元素放入队列，放入队列才是目的。当队列满时会阻塞，直到队列空出来位置再放入。
 *  E take() throws InterruptedException; // 本意是从队列取出第一个元素，取出元素才是目的。当队列为空时会阻塞，直到队列有元素时再取出。

#### 举例 ####

ArrayBlockQueue是基于一把ReentrantLock实现了各操作的线程安全，保证了只有一个线程访问临界区。正因为如此，通过它，我们可以重写一下前面讲的生产消费模式。[通过管程和信号量分别实现生产者消费者模式][Link 1]

package juc;
    
    import org.junit.Test;
    
    import java.util.concurrent.ArrayBlockingQueue;
    import java.util.concurrent.BlockingQueue;
    
    public class ArrayBlockingQueueTest { 
    
    	class Consumer implements Runnable { 
    
    		private BlockingQueue blockingQueue;
    
    		Consumer(BlockingQueue blockingQueue) { 
    			this.blockingQueue = blockingQueue;
    		}
    
    		@Override
    		public void run() { 
    			for (; ; ) { 
    				System.out.println(Thread.currentThread().getName() + "消费,当前容量:" + blockingQueue.size());
    				try { 
    					Thread.sleep(1000);
    					blockingQueue.take();
    				} catch (InterruptedException e) { 
    					e.printStackTrace();
    				}
    			}
    		}
    	}
    
    	class Producer implements Runnable { 
    
    		private BlockingQueue blockingQueue;
    
    		Producer(BlockingQueue blockingQueue) { 
    			this.blockingQueue = blockingQueue;
    		}
    
    		@Override
    		public void run() { 
    			for (; ; ) { 
    				System.out.println(Thread.currentThread().getName() + "生产,当前容量:" + blockingQueue.size());
    				try { 
    					Thread.sleep(1000);
    					blockingQueue.put("product");
    				} catch (InterruptedException e) { 
    					e.printStackTrace();
    				}
    			}
    		}
    	}
    
    
    	@Test
    	public void test() throws InterruptedException { 
    		final BlockingQueue blockingQueue = new ArrayBlockingQueue(10);
    		// 新建两个生产者线程+一个消费者线程，（生产是比消费快的，当缓冲区满时，生产者会阻塞）
    		Thread t1 = new Thread(new Producer(blockingQueue), "生产者1");
    		Thread t2 = new Thread(new Producer(blockingQueue), "生产者2");
    		Thread t3 = new Thread(new Consumer(blockingQueue), "消费者1");
    		t1.start();
    		t2.start();
    		t3.start();
    		t1.join();
    		t2.join();
    		t3.join();
    	}
    }

###### 执行结果 ######

![在这里插入图片描述][watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNzk2ODYxMw_size_16_color_FFFFFF_t_70_pic_center]  
这是执行稳定之后的日志，因为生产者有两个线程，消费者只有一个，所有共享区的元素个数保持在最大容量。

#### 源码解读 ####

前面我们讲了，阻塞队列有着并发安全的阻塞读和阻塞写，对应的就是blockQueue的take和put方法。下面我们针对这两个方法，看看是怎么实现的。

*  首先存储对象用的是数组，并记录了下一个take的索引，下一个put的索引位置，队列大小count。

/** The queued items */
        final Object[] items;
    
        /** items index for next take, poll, peek or remove */
        int takeIndex;
    
        /** items index for next put, offer, or add */
        int putIndex;
    
        /** Number of elements in the queue */
        int count;

*  通过一个锁和两个条件队列实现队列的阻塞读写和并发安全。

/** Main lock guarding all access */
        final ReentrantLock lock;
    
        /** Condition for waiting takes */
        private final Condition notEmpty;
    
        /** Condition for waiting puts */
        private final Condition notFull;

*  看take方法，阻塞读：

public E take() throws InterruptedException { 
        	// 获取锁，加锁
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly();
            try { 
            	// 队列为空，读线程在notEmpty条件队列里阻塞
                while (count == 0)
                    notEmpty.await();
                // 队列不空，就取出一个元素，前面提到的takeIndex就起作用了
                return dequeue();
            } finally { 
            	// 取完之后释放锁
                lock.unlock();
            }
        }
    
        private E dequeue() { 
            final Object[] items = this.items;
    		// 取出takeIndex元素
            E x = (E) items[takeIndex];
            // 取出后置为null
            items[takeIndex] = null;
            // 如果取出的元素是队列的最后一个，takeIndex移到数组开头。循环队列
            if (++takeIndex == items.length)
                takeIndex = 0;
            // 数据量 -1
            count--;
            // 维护一个迭代器 
            if (itrs != null)
                itrs.elementDequeued();
            // 出队列之后，唤醒条件队列notFull的一个线程，可以入队 
            notFull.signal();
            return x;
        }

看完take方法相对还是比较清晰，主要是通过锁和条件队列实现了阻塞读的功能。  
下面看下put阻塞写方法：

public void put(E e) throws InterruptedException { 
            checkNotNull(e);
            final ReentrantLock lock = this.lock;
            lock.lockInterruptibly();
            try { 
            	// 当元素数量与容量相等，写线程将在notFull条件队列里排队，直到take方法的notFull.signal唤醒
                while (count == items.length)
                    notFull.await();
                // 执行入队操作
                enqueue(e);
            } finally { 
            	// 写完之后释放锁
                lock.unlock();
            }
        }
    
    	    private void enqueue(E x) { 
            final Object[] items = this.items;
            // 将入队的元素放在putIndex的位置
            items[putIndex] = x;
            // 如果putIndex为最后一个，则从头开始放。（这里会不会出现putIndex与takeIndex位置发生错乱的问题，不会，可以通过count进行控制）
            if (++putIndex == items.length)
                putIndex = 0;
            count++;
            // 唤醒notEmpty条件队列的线程，可以入队
            notEmpty.signal();
        }

#### 小结 ####

1、BlockingQueue的功能其实是实现了一个线程安全的生产者消费者模式。  
2、BlockingQueue也常用在线程池的等待队列中，如ArrayBlockingQueue和LinkedBlockingQueue

[Link 1]: https://blog.csdn.net/weixin_37968613/article/details/106303293?ops_request_misc=%257B%2522request%255Fid%2522%253A%2522160522766219725225046478%2522%252C%2522scm%2522%253A%252220140713.130102334.pc%255Fblog.%2522%257D&request_id=160522766219725225046478&biz_id=0&utm_medium=distribute.pc_search_result.none-task-blog-2~blog~first_rank_v2~rank_blog_default-1-106303293.pc_v2_rank_blog_default&utm_term=%E7%94%9F%E4%BA%A7&spm=1018.2118.3001.4450
[watermark_type_ZmFuZ3poZW5naGVpdGk_shadow_10_text_aHR0cHM6Ly9ibG9nLmNzZG4ubmV0L3dlaXhpbl8zNzk2ODYxMw_size_16_color_FFFFFF_t_70_pic_center]: /images/20210518/8fe174317e2b4f7e80f5e6d370a814af.png