今天要讲的是ArrayBlockQueue,ArrayBlockQueue是JUC提供的线程安全的有界的阻塞队列,一看到Array,第一反应:这货肯定和数组有关,既然是数组,那自然是有界的了,我们先来看看ArrayBlockQueue的基本使用方法,然后再看看ArrayBlockQueue的源码。
ArrayBlockQueue基本使用
public static void main(String[] args) throws InterruptedException { ArrayBlockingQueue<Integer> arrayBlockingQueue=new ArrayBlockingQueue(5); arrayBlockingQueue.offer(10); arrayBlockingQueue.offer(50); arrayBlockingQueue.add(20); arrayBlockingQueue.add(60); System.out.println(arrayBlockingQueue); System.out.println(arrayBlockingQueue.poll()); System.out.println(arrayBlockingQueue); System.out.println(arrayBlockingQueue.take()); System.out.println(arrayBlockingQueue); System.out.println(arrayBlockingQueue.peek()); System.out.println(arrayBlockingQueue); }
运行结果:
代码比较简单,但是你肯定会有疑问
要解决上面几个疑问,最好的办法当然是看下源码,通过亲自阅读源码所产生的印象远远要比看视频,看博客,死记硬背最后的结论要深刻的多。就算真的忘记了,只要再看看源码,瞬间可以回忆起来。
ArrayBlockQueue源码解析
构造方法
ArrayBlockQueue提供了三个构造方法,如下图所示:
ArrayBlockingQueue(int capacity)
public ArrayBlockingQueue(int capacity) { this(capacity, false); }
这是最常用的构造方法,传入capacity,capacity是容量的意思,也就是ArrayBlockingQueue的最大长度,方法内部直接调用了第二个构造方法,传入的第二个参数为false。
ArrayBlockingQueue(int capacity, boolean fair)
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,fair是boolean类型的,代表是公平锁,还是非公平锁,可以看出如果我们用第一个构造方法来创建ArrayBlockingQueue的话,采用的是非公平锁,因为公平锁会损失一定的性能,在没有充足的理由的情况下,是没有必要采用公平锁的。
方法内部做了几件事情:
至于排他锁和两个条件变量是做什么用的,看到后面就明白了。
ArrayBlockingQueue(int capacity, boolean fair,Collection<? extends E> c)
public ArrayBlockingQueue(int capacity, boolean fair, Collection<? extends E> c) { //调用第二个构造方法,方法内部就是初始化数组,排他锁,两个条件变量 this(capacity, fair); final ReentrantLock lock = this.lock; lock.lock(); // 开启排他锁 try { int i = 0; try { // 循环传入的集合,把集合中的元素赋值给items数组,其中i会自增 for (E e : c) { checkNotNull(e); items[i++] = e; } } catch (ArrayIndexOutOfBoundsException ex) { throw new IllegalArgumentException(); } count = i;//把i赋值给count //如果i==capacity,也就是到了最大容量,把0赋值给putIndex,否则把i赋值给putIndex putIndex = (i == capacity) ? 0 : i; } finally { lock.unlock();//释放排他锁 } }
看到这里,我们应该明白这个构造方法的作用是什么了,就是把传入的集合作为ArrayBlockingQueuede初始化数据,但是我们又会有一个新的疑问:count,putIndex 是做什么用的。
offer(E e)
public boolean offer(E e) { checkNotNull(e); final ReentrantLock lock = this.lock; lock.lock();//开启排他锁 try { if (count == items.length)//如果count==items.length,返回false return false; else { enqueue(e);//入队 return true;//返回true } } finally { lock.unlock();//释放锁 } }
看到这里,我们应该可以明白了,ArrayBlockQueue是如何保证线程安全的,还是利用了ReentrantLock排他锁,count就是用来保存数组的当前大小的。我们再来看看enqueue方法。
private void enqueue(E x) { final Object[] items = this.items; items[putIndex] = x; if (++putIndex == items.length) putIndex = 0; count++; notEmpty.signal(); }
这方法比较简单,在代码里面就不写注释了,做了如下的操作:
这里就解答了一个疑问:putIndex是做什么的,就是入队元素的下标。
add(E e)
public boolean add(E e) { return super.add(e); }
public boolean add(E e) { if (offer(e)) return true; else throw new IllegalStateException("Queue full"); }
这个方法内部最终还是调用的offer方法。
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 notFull.await(); enqueue(e);//入队 } finally { lock.unlock();//释放排他锁 } }
可以看到put方法和 offer/add方法的区别了:
poll()
public E poll() { final ReentrantLock lock = this.lock; lock.lock(); try { return (count == 0) ? null : dequeue(); } finally { lock.unlock(); } }
我们来看dequeue方法:
private E dequeue() { final Object[] items = this.items; @SuppressWarnings("unchecked") E x = (E) items[takeIndex];//获得元素的值 items[takeIndex] = null;//把null赋值给items[takeIndex] if (++takeIndex == items.length)//如果takeIndex自增后的值== items.length,就把0赋值给takeIndex takeIndex = 0; count--; if (itrs != null) itrs.elementDequeued(); notFull.signal();//唤醒因为调用notFull的await方法而被阻塞的线程 return x; }
这里调用了notFull的signal方法来唤醒因为调用notFull的await方法而被阻塞的线程,那到底在哪里调用了notFull的await方法呢,还记不记得在put方法中调用了notFull的await方法,我们再看看:
while (count == items.length) notFull.await();
当队列满了,就调用 notFull.await()来等待,在出队操作中,又调用了notFull.signal()来唤醒。
take()
public E take() throws InterruptedException { final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try { while (count == 0) notEmpty.await(); return dequeue(); } finally { lock.unlock(); } }
这里调用了notEmpty的await方法,那么哪里调用了notEmpty的signal方法呢?在enqueue入队方法里。
我们可以看到take和poll的区别:
peek()
public E peek() { final ReentrantLock lock = this.lock; lock.lock(); try { return itemAt(takeIndex); } finally { lock.unlock(); } }
final E itemAt(int i) { return (E) items[i]; }
我们可以看到peek和poll/take的区别:
size()
public int size() { final ReentrantLock lock = this.lock; lock.lock(); try { return count; } finally { lock.unlock(); } }
总结
至此,ArrayBlockQueue的核心源码就分析完毕了,我们来做一个总结:
以上所述是小编给大家介绍的ArrayBlockQueue源码解析详解整合,希望对大家有所帮助,如果大家有任何疑问请给我留言,小编会及时回复大家的。在此也非常感谢大家对亿速云网站的支持!
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。