Java的多线程之阻塞队列

一、阻塞队列

之前已经初步了解了Java并发程序设计基础的底层构建块。然而，对于实际编程来说，应该尽可能远离底层结构。使用并发处理的专业人士实现的较高层次的结构要方便的多、安全的多。

对于许多线程问题，可以通过使用一个或多个队列以优雅且安全的方式将其形式化。生产者队列向队列插入元素，消费者队列取出它们。使用队列，可以安全地从一个线程向另一个线程传递数据。

当试图向队列添加元素而队列已满，或是想从队列移出元素而队列为空的时候，阻塞队列（blocking queue）导致线程阻塞。在协调多个线程之间的合作时，阻塞队列是一个有用的工具。工作者线程可以周期性将中间结果存储在阻塞队列中，其他的工作者线程移出中间结果并进一步加以修改。队列会自动地平衡负载。



阻塞队列方法分为以下3类，这取决于当队列满或空时它们的响应方式。如果将队列当作线程管理工具来使用，将用到put和take方法。当试图向满队列中添加或从空的队列中移出元素时，add、remove和element操作抛出异常。当然，在一个多线程程序中，队列会在任何时候空或满，因此，一定要使用offer、poll和peek方法作为替代。这样只会给出错误提示而不会抛出异常。

注：poll和peek方法返回空来指示失败。因此，向这些队列中插入null值时非法的。

java.util.concurrent包提供了阻塞队列的几个变种。默认情况下，LinkedBlockingQueue的容量是没有上边界的，但是，也可以选择指定最大容量。LinkedBolckingDeque是一个双端的版本。ArrayBlockingQueue在构造时需要指定容量，并且有一个可选的参数来指定是否需要公平性。若设置公平性，则等待最长时间的线程会优先得到处理。通常，公平性会降低性能，只有在确定非常需要时才使用它。

PriorityBlockingQueue是一个带优先级的队列，而不是先进先出队列。

Java SE7增加了一个TransferQueue接口，允许生产者线程等待，直到消费者准备就绪可以接收一个元素。生产者调用q.transfer(item)会阻塞，直到另一个线程将元素删除。LinkedTransferQueue类实现了这个接口。

例如：

package com.thread.blockingQueue;

import java.io.File;
import java.util.Scanner;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

public class BlockingQueueTest {

public static void main(String[] args) {
Scanner in = new Scanner(System.in);
System.out.print("Enter base directory:");
String directory = in.nextLine();
System.out.print("Enter keyword:");
String keyword = in.nextLine();
final int FILE_QUEUE_SIZE = 10;
final int SEARCH_THREADS = 100;

BlockingQueue<File> queue = new ArrayBlockingQueue<>(FILE_QUEUE_SIZE);

FileEnumerationTask enumerator = new FileEnumerationTask(queue,
new File(directory));
new Thread(enumerator).start();
for (int i = 1; i <= SEARCH_THREADS; i++) {
new Thread(new SearchTask(queue, keyword)).start();
}
}

}

package com.thread.blockingQueue;

import java.io.File;
import java.util.concurrent.BlockingQueue;

public class FileEnumerationTask implements Runnable {

public static File DUMMY = new File("");
private BlockingQueue<File> queue;
private File startingDirectory;

public FileEnumerationTask(BlockingQueue<File> queue, File startingDirectory) {
this.queue = queue;
this.startingDirectory = startingDirectory;
}

@Override
public void run() {
try {
enumerate(startingDirectory);
queue.put(DUMMY);
} catch (InterruptedException e) {
e.printStackTrace();
}
}

public void enumerate(File directory) throws InterruptedException {
File[] files = directory.listFiles();
for (File file : files) {
if (file.isDirectory()) {
System.out.println(file.getAbsolutePath());
enumerate(file);
} else
queue.put(file);
}
}

}

package com.thread.blockingQueue;

import java.io.File;
import java.io.FileNotFoundException;
import java.util.Scanner;
import java.util.concurrent.BlockingQueue;

public class SearchTask implements Runnable {

private BlockingQueue<File> queue;
private String keyword;

public SearchTask(BlockingQueue<File> queue, String keyword) {
this.queue = queue;
this.keyword = keyword;
}

@Override
public void run() {
try {
boolean done = false;
while (!done) {
File file = queue.take();
if (file == FileEnumerationTask.DUMMY) {
queue.put(file);
done = true;
} else
search(file);
}
} catch (InterruptedException e) {
e.printStackTrace();
} catch (FileNotFoundException e) {
e.printStackTrace();
}
}

public void search(File file) throws FileNotFoundException {
try (Scanner in = new Scanner(file)) {
int lineNumber = 0;
while (in.hasNextLine()) {
lineNumber++;
String line = in.nextLine();
if (line.contains(keyword)) {
System.out.printf("%s:%d:%s%n", file.getPath(), lineNumber,
line);
}
}
}
}
}

生产者线程枚举在所有子目录下的所有文件并把它们放在一个阻塞队列中。这个操作很快，如果没有上限的话，很快就包含了所有找到的文件。

我们同时启动了大量搜索线程。每个搜索线程从队列中取出一个文件，打开它，打印所有包含该关键字的行，然后取出下一个文件。我们使用一个小技巧在工作结束后终止这个应用程序。为了发出完成信号，枚举线程放置一个虚拟对象到队列中。当搜索线程取到这个虚拟对象时，将其放回并终止。

注意，不需要显式的线程同步。在这个应用程序中，我们使用队列数据结构作为一种同步机制。

常用方法：ArrayBlockingQueue(int capacity) 构造一个带有指定容量和公平性设置的阻塞队列。该队列使用循环数组实现。

LinkedBlockingQueue()

LinkedBlockingDeque() 构造一个无上限的阻塞队列或双向队列，用链表实现。