BlockingQueue
BlockingQueue
- 无限队列（unbounded queue）
: 几乎可以无限增长 - 有限队列（bounded queue）
: 定义了最大容量
无限队列
创建一个无限队列的方法很简单：
BlockingQueue<String> blockingQueue = new LinkedBlockingDeque<>();
上面这段代码中，
有限队列
第二种类型的队列是有限队列。我们可以通过将容量作为参数传递给构造函数来创建这样的队列
BlockingQueue<String> blockingQueue = new LinkedBlockingDeque<>(10);
上面这句代码中，我们设置了
使用有限队列是设计并发程序的好方法，因为当我们将元素插入到已经满了的队列时，这些操作需要等到消费者赶上并在队列中提供一些空间。这种机制可以让那个我们不做任何其它更改就可以实现节流。
BlockingQueue API
添加元素
| 方法 | 说明 |
|---|---|
| add() | 如果插入成功则返回 |
| put() | 将指定的元素插入队列，如果队列满了，那么会阻塞直到有空间插入 |
| offer() | 如果插入成功则返回 |
| offer(E e, long timeout, TimeUnit unit) | 尝试将元素插入队列，如果队列已满，那么会阻塞直到有空间插入 |
检索元素
| 方法 | 说明 |
|---|---|
| take() | 获取队列的头部元素并将其删除，如果队列为空，则阻塞并等待元素变为可用 |
| poll(long timeout, TimeUnit unit) | 检索并删除队列的头部，如有必要，等待指定的等待时间以使元素可用，如果超时，则返回 |
在构建生产者
多线程生产者- 消费者示例
接下来我们创建一个由两部分组成的程序
public class NumbersProducer implements Runnable {
private BlockingQueue<Integer> numbersQueue;
private final int poisonPill;
private final int poisonPillPerProducer;
public NumbersProducer(BlockingQueue<Integer> numbersQueue, int poisonPill, int poisonPillPerProducer) {
this.numbersQueue = numbersQueue;
this.poisonPill = poisonPill;
this.poisonPillPerProducer = poisonPillPerProducer;
}
public void run() {
try {
generateNumbers();
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
private void generateNumbers() throws InterruptedException {
for (int i = 0; i < 100; i++) {
numbersQueue.put(ThreadLocalRandom.current().nextInt(100));
}
for (int j = 0; j < poisonPillPerProducer; j++) {
numbersQueue.put(poisonPill);
}
}
}
我们的生成器构造函数将
public class NumbersConsumer implements Runnable {
private BlockingQueue<Integer> queue;
private final int poisonPill;
public NumbersConsumer(BlockingQueue<Integer> queue, int poisonPill) {
this.queue = queue;
this.poisonPill = poisonPill;
}
public void run() {
try {
while (true) {
Integer number = queue.take();
if (number.equals(poisonPill)) {
return;
}
System.out.println(Thread.currentThread().getName() + " result: " + number);
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
}
}
}
需要注意的重要事项是队列的使用。与生成器构造函数中的相同，队列作为参数传递。我们可以这样做，是因为
int BOUND = 10;
int N_PRODUCERS = 4;
int N_CONSUMERS = Runtime.getRuntime().availableProcessors();
int poisonPill = Integer.MAX_VALUE;
int poisonPillPerProducer = N_CONSUMERS / N_PRODUCERS;
int mod = N_CONSUMERS % N_PRODUCERS;
BlockingQueue<Integer> queue = new LinkedBlockingQueue<>(BOUND);
for (int i = 1; i < N_PRODUCERS; i++) {
new Thread(new NumbersProducer(queue, poisonPill, poisonPillPerProducer)).start();
}
for (int j = 0; j < N_CONSUMERS; j++) {
new Thread(new NumbersConsumer(queue, poisonPill)).start();
}
new Thread(new NumbersProducer(queue, poisonPill, poisonPillPerProducer + mod)).start();
当我们运行程序时，