阻塞队列是Java5线程新特征中的内容,Java定义了阻塞队列的接口java.util.concurrent.BlockingQueue,阻塞队列的概念是,一个指定长度的队列,如果队列满了,添加新元素的操作会被阻塞等待,直到有空位为止。同样,当队列为空时候,请求队列元素的操作同样会阻塞等待,直到有可用元素为止。
有界队列。当使用有限的 maximumPoolSizes 时,有界队列(如 ArrayBlockingQueue)有助于防止资源耗尽,但是可能较难调整和控制。队列大小和最大池大小可能需要相互折衷:使用大型队列和小型池可以最大限度地降低 CPU 使用率、操作系统资源和上下文切换开销,但是可能导致人工降低吞吐量。如果任务频繁阻塞(例如,如果它们是 I/O 边界),则系统可能为超过您许可的更多线程安排时间。使用小型队列通常要求较大的池大小,CPU 使用率较高,但是可能遇到不可接受的调度开销,这样也会降低吞吐量。
条件变量 lock: http://my.oschina.net/jielucky/blog/156827
package com.thread.q;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
public class QueueTest {
private final BlockingQueue<Object> queue = new ArrayBlockingQueue<Object>(
2);
// {
// try {
// for (int i = 0; i <= 0; i++)
// queue.put(1);
// } catch (InterruptedException e) {
// e.printStackTrace();
// }
// }
public void test() throws InterruptedException, ExecutionException {
System.out.println("队列里总共有:" + queue.size());
// 创建一个线程池
ExecutorService pool = Executors.newFixedThreadPool(2);
// 创建两个有返回值的任务
Callable<String> readThread = new ReadThread(queue);
Callable<String> writeThread = new WriteThread(queue);
// 执行任务并获取Future对象
Future<String> writeFuture = pool.submit(writeThread);
Future<String> readFuture = pool.submit(readThread);
// 执行
// writeFuture.get();
writeFuture.get();
readFuture.get();
System.out.println("队列里还有:" + queue.size());
pool.shutdownNow();
}
public static void main(String[] args) {
try {
new QueueTest().test();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}
}
class ReadThread implements Callable<String> {
BlockingQueue<Object> queue;
ReadThread(BlockingQueue<Object> queue) {
this.queue = queue;
}
@Override
public String call() throws Exception {
for (int i = 0; i < 9; i++) {
Thread.sleep(1000);
// 获取并移除此队列的头,如果此队列为空,则返回 null
Object obj = queue.poll();
if (obj == null) {
continue;
}
System.out.println("读:队列里还有 " + queue.size());
}
// return "读:队列里还有" + queue.size();
return "";
}
}
class WriteThread implements Callable<String> {
BlockingQueue<Object> queue;
WriteThread(BlockingQueue<Object> queue) {
this.queue = queue;
}
@Override
public String call() throws Exception {
// 将指定的元素插入此队列的尾部,如果该队列已满,则等待可用的空间。
for (int i = 1; i < 10; i++) {
queue.put(i);
System.out.println("写:队列里还有" + queue.size());
}
// return "写:队列里还有" + queue.size();
return "";
}
}