线程阻塞队列ArrayBlockingQueue

原创
2013/08/23 17:40
阅读数 1.5K

     阻塞队列是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 "";
	}
}
展开阅读全文
加载中
点击引领话题📣 发布并加入讨论🔥
打赏
0 评论
6 收藏
0
分享
返回顶部
顶部