本文共 7406 字，大约阅读时间需要 24 分钟。
转自：java并发编程实战
BlockingQueue阻塞队列提供可阻塞的put和take方法，以及支持定时的offer和poll方法。如果队列已经满了，那么put方法将阻塞直到空间可用；如果队列为空，那么take方法将阻塞直到有元素可用。队列可以是有界的也可以是无界的。
如果生产者生成工作的速率比消费者处理工作的速率款，那么工作项会在队列中累计起来，最终好紧内存。同样，put方法的阻塞特性也极大地简化了生产者的编码。如果使用有界队列，当队列充满时，生产者将阻塞并不能继续生产工作，而消费者就有时间来赶上工作的进度。阻塞队列同样提供了一个offer方法，如果数据项不能被添加到队列中，那么将返回一个失败的状态。这样你就能创建更多灵活的策略来处理负荷过载的情况。
在构建高可靠的应用程序时，有界队列是一种强大的资源管理工具：他们能一直并防止产生过多的工作项，使应用程序在负荷过载的情况下边的更加健壮。
/** * java并发编程实战 * 5.3.1桌面搜索 * 爬虫查找所有文件并放入队列 * Created by mrf on 2016/3/7. */public class FileCrawler implements Runnable { private final BlockingQueue fileQueue; private final FileFilter fileFilter; private final File root; public FileCrawler(BlockingQueue fileQueue, FileFilter fileFilter, File root) { this.fileQueue = fileQueue; this.fileFilter = fileFilter; this.root = root; } @Override public void run() { try { crawl(root); } catch (InterruptedException e) { //恢复中断 Thread.currentThread().interrupt(); e.printStackTrace(); } } private void crawl(File root) throws InterruptedException { File[] entries = root.listFiles(fileFilter); if (entries!=null){ for (File entry : entries) { if (entry.isDirectory()){ crawl(entry); }else if (!alreadyIndexed(entry)){ fileQueue.put(entry); } } } } private boolean alreadyIndexed(File entry){ //检查是否加入索引 return false; }}/** * 消费者 * 将爬虫结果队列取出并加入索引 */class Indexer implements Runnable{ private static final int BOUND = 100; private static final int N_CONSUMERS = 2; private final BlockingQueue queue; Indexer(BlockingQueue queue) { this.queue = queue; } @Override public void run() { try { while (true){ indexFile(queue.take()); } }catch (InterruptedException e){ Thread.currentThread().interrupt(); } } private void indexFile(File take) { //将文件加入索引 } public static void startIndexing(File[] roots){ BlockingQueue queue = new LinkedBlockingDeque<>(BOUND); FileFilter fileFilter = new FileFilter() { @Override public boolean accept(File pathname) { return true; } }; for (File root:roots) { new Thread(new FileCrawler(queue,fileFilter,root)).start(); } for (int i = 0; i < N_CONSUMERS; i++) { new Thread(new Indexer(queue)).start(); } }} Semaphore中管理着一组虚拟的许可（permit）。许可的初始数量可通过构造函数来指定。在执行操作时可以首先获得许可（只要还有剩余的许可），并在使用以后释放许可。如果没有许可，那么acquire将阻塞直到有许可（或者被中断或者操作超时）。release方法将返回一个许可给信号量。计算信号量的一种简化形式是二值信号量，即初始值为1的Semaphore。二值信号量可以用作互斥体（mutex），并具备不可重入的加锁语义：谁拥有这个唯一的许可，谁就拥有了互斥锁。
/** * java 并发编程实战 * 5-14使用Semaphore做容器设置边界 * 信号量 * Created by mrf on 2016/3/8. */public class BoundedHashSet{ private final Set set; private final Semaphore sem;// public BoundedHashSet(Set set, Semaphore sem) {// this.set = set;// this.sem = sem;// } public BoundedHashSet(int bound){ this.set = Collections.synchronizedSet(new HashSet ()); sem = new Semaphore(bound); } public boolean add(T o) throws InterruptedException { sem.acquire(); boolean wasAdded = false; try { wasAdded = set.add(o); return wasAdded; }finally { if (!wasAdded){ sem.release(); } } } public boolean remove(Object o){ boolean wasRemoved = set.remove(o); if (wasRemoved){ sem.release(); } return wasRemoved; }}