有三个构造函数,必须设定 队列的大小, 公平和非公平可选。默认情况下不保证线程公平的访问队列,所谓公平访问队列是指阻塞的线程,可以按照阻塞的先后顺序访问队列,即先阻塞线程先访问队列。对于元素而言是FIFO的原则。
构造函数1
public ArrayBlockingQueue(int capacity) { this(capacity, false); // 默认非公平 }
构造函数2
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(); }
构造函数3
设定从集合中初始化队列,
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 { for (E e : c) { checkNotNull(e); // 保证加入的元素不为空 items[i++] = e; } } catch (ArrayIndexOutOfBoundsException ex) { throw new IllegalArgumentException(); } count = i; putIndex = (i == capacity) ? 0 : i; } finally { lock.unlock(); } }
Offer 和 Add
add方法调用offer,如果添加失败,则抛出异常。
public boolean offer(E e) { checkNotNull(e); final ReentrantLock lock = this.lock; lock.lock(); // 加锁 try { if (count == items.length) // 如果队列满了,添加失败 return false; else { enqueue(e); return true; } } finally { lock.unlock(); } }
其中 enqueue 方法的如下:
private void enqueue(E x) { // assert lock.getHoldCount() == 1; // assert items[putIndex] == null; final Object[] items = this.items; items[putIndex] = x; // 在putIndex位置存放元素 if (++putIndex == items.length) // 更新putindex位置 putIndex = 0; count++; notEmpty.signal(); // 通知挂载在notEmpty上的线程,去消费。 }
Put方法
put()方法添加如果不成功则会阻塞。
public void put(E e) throws InterruptedException { checkNotNull(e); final ReentrantLock lock = this.lock; lock.lockInterruptibly(); // 加可中断的锁 try { while (count == items.length) // 如果队列满 notFull.await(); // 释放锁,挂载到notFull条件的等待队列上 enqueue(e); // 入队列 } finally { lock.unlock(); } }
还有另外一个offer方法,等待特定时间
public boolean offer(E e, long timeout, TimeUnit unit) throws InterruptedException { checkNotNull(e); long nanos = unit.toNanos(timeout); final ReentrantLock lock = this.lock; lock.lockInterruptibly(); try { while (count == items.length) { if (nanos <= 0) return false; nanos = notFull.awaitNanos(nanos); // 等待特定的纳秒直到超时,则唤醒该线程 } enqueue(e); return true; } finally { lock.unlock(); } }
poll()方法
public E poll() { final ReentrantLock lock = this.lock; lock.lock(); try { return (count == 0) ? null : dequeue(); // 退出队列 } finally { lock.unlock(); } }
其中dequeue方法具体如下:
private E dequeue() { // assert lock.getHoldCount() == 1; // assert items[takeIndex] != null; final Object[] items = this.items; @SuppressWarnings("unchecked") E x = (E) items[takeIndex]; items[takeIndex] = null; // 指定取出的位置元素为null if (++takeIndex == items.length) // 取index更新 takeIndex = 0; count--; if (itrs != null) // 将所有迭代器中的该元素删除 itrs.elementDequeued(); notFull.signal(); // 通知挂在notFull上的等待线程取获取锁。 return x; }
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(); } }
对于阻塞队列通常提供的方法实现的语义:
方法/处理方式 | 抛出异常 | 返回特殊值 | 一直阻塞 | 超时退出 |
---|---|---|---|---|
插入方法 | add(e) | offer(e) | put(e) | offer(e,time,unit) |
移除方法 | remove() | poll() | take() | poll(time, unit) |
检擦方法 | element() | peek() | 不可用 | 不可用 |