Java阻塞队列
程序员文章站
2024-02-11 12:45:28
...
管程
管程 (英语:Moniters,也称为监视器) 是一种程序结构,结构内的多个子程序(对象或模块)形成的多个工作线程互斥访问共享资源。
管程实现了在一个时间点,最多只有一个线程在执行管程的某个子程序。
管程包括:
- 多个彼此可以交互并共用资源的线程
- 多个与资源使用有关的变量
- 一个互斥锁
- 一个用来避免竞态条件的不变量
信号量
阻塞队列
阻塞队列(BlockingQueue)是一个支持两个附加操作的队列。
在队列为空时,获取元素的线程会等待队列变为非空。
当队列满时,存储元素的线程会等待队列可用。
阻塞队列常用于生产者和消费者的场景,生产者是往队列里添加元素的线程,消费者是从队列里拿元素的线程。
阻塞队列就是生产者存放元素的容器,而消费者也只从容器里拿元素。
Queue接口
Queue接口的三种操作:入队、出队和检索均有两个实现
- 入队:add() offer()
- 出队:remove() poll()
- 检索:element() peek()
BlockingQueue
BlockingQueue在Queue接口的基础上对入队和出队两个操作分别又增加了阻塞方法
- 阻塞入队 put(E e)
如果BlockQueue没有空间,则调用此方法的线程被阻断 - 阻塞出队 E take()
取走BlockingQueue里排在首位的对象,若BlockingQueue为空,阻断进入等待状态直到BlockingQueue有新的数据被加入; - 定时入队
- 定时出队
ArrayBlockingQueue
ArrayBlockingQueue是一个用数组实现的有界阻塞队列。
按照先进先出(FIFO)的原则对元素进行排序。
ArrayBlockingQueue在生产者放入数据和消费者获取数据,都是共用同一个锁对象
重要对象:
final Object[] items;
int takeIndex;
int putIndex;
int count;
final ReentrantLock lock;
private final Condition notEmpty;
private final Condition notFull;
无参构造函数 默认非公平
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();
}
访问者的公平性是使用可重入锁实现的
LinkedBlockingQueue
- LinkedBlockingQueue是一个用链表实现的有界阻塞队列。
- 此队列的默认和最大长度为Integer.MAX_VALUE。
public LinkedBlockingQueue() {
this(Integer.MAX_VALUE);
}
- 此队列按照先进先出的原则对元素进行排序。
PriorityBlockingQueue
- PriorityBlockingQueue是一个支持优先级的*队列。
- 默认情况下元素采取自然顺序排列,也可以通过比较器comparator来指定元素的排序规则。
- 元素按照升序排列。
LinkedBlockingDeque
LinkedBlockingDeque: 一个基于双端链表的双端阻塞队列
阻塞队列的实现原理
让生产者和消费者能够高效率的进行通讯呢?
使用通知模式实现。
当生产者往满的队列里添加元素时会阻塞住生产者,当消费者消费了一个队列中的元素后,会通知生产者当前队列可用。
ArrayBlockingQueue使用了Condition来实现
public void put(E e) throws InterruptedException {
checkNotNull(e);
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == items.length)
notFull.await();
enqueue(e);
} finally {
lock.unlock();
}
}
public E take() throws InterruptedException {
final ReentrantLock lock = this.lock;
lock.lockInterruptibly();
try {
while (count == 0)
notEmpty.await();
return dequeue();
} finally {
lock.unlock();
}
}
private void enqueue(E x) {
// assert lock.getHoldCount() == 1;
// assert items[putIndex] == null;
final Object[] items = this.items;
items[putIndex] = x;
if (++putIndex == items.length)
putIndex = 0;
count++;
notEmpty.signal();
}
上一篇: 机器学习实战—决策树
下一篇: Java 谈谈阻塞队列