Java阻塞队列线程集控制的实现
2014-12-26 00:00
741 查看
队列以一种先进先出的方式管理数据。如果你试图向一个已经满了的阻塞队列中添加一个元素,或是从一个空的阻塞队列中移除一个元素,将导致线程阻塞。在多线程进行合作时,阻塞队列是很有用的工具。工作者线程可以定期的把中间结果存到阻塞队列中。而其他工作者线程把中间结果取出并在将来修改它们。队列会自动平衡负载。如果第一个线程集运行的比第二个慢,则第二个线程集在等待结果时就会阻塞。如果第一个线程集运行的快,那么它将等待第二个线程集赶上来。
下面的程序展示了如何使用阻塞队列来控制线程集。程序在一个目录及它的所有子目录下搜索所有文件,打印出包含指定关键字的文件列表。
java.util.concurrent包提供了阻塞队列的4个变种:LinkedBlockingQueue、ArrayBlockingQueue、PriorityBlockingQueue和DelayQueue。我们用的是ArrayBlockingQueue。ArrayBlockingQueue在构造时需要给定容量,并可以选择是否需要公平性。如果公平参数被设置了,等待时间最长的线程会优先得到处理。通常,公平性会使你在性能上付出代价,只有在的确非常需要的时候再使用它。
生产者线程枚举在所有子目录下的所有文件并把它们放到一个阻塞队列中。这个操作很快,如果队列没有设上限的话,很快它就包含了没有找到的文件。
我们同时还启动了大量的搜索线程。每个搜索线程从队列中取出一个文件,打开它,打印出包含关键字的所有行,然后取出下一个文件。我们使用了一个小技巧来在工作结束后终止线程。为了发出完成信号,枚举线程把一个虚拟对象放入队列。(这类似于在行李输送带上放一个写着“最后一个包”的虚拟包。)当搜索线程取到这个虚拟对象时,就将其放回并终止。
注意,这里不需要人任何显示的线程同步。在这个程序中,我们使用队列数据结构作为一种同步机制。
下面的程序展示了如何使用阻塞队列来控制线程集。程序在一个目录及它的所有子目录下搜索所有文件,打印出包含指定关键字的文件列表。
java.util.concurrent包提供了阻塞队列的4个变种:LinkedBlockingQueue、ArrayBlockingQueue、PriorityBlockingQueue和DelayQueue。我们用的是ArrayBlockingQueue。ArrayBlockingQueue在构造时需要给定容量,并可以选择是否需要公平性。如果公平参数被设置了,等待时间最长的线程会优先得到处理。通常,公平性会使你在性能上付出代价,只有在的确非常需要的时候再使用它。
生产者线程枚举在所有子目录下的所有文件并把它们放到一个阻塞队列中。这个操作很快,如果队列没有设上限的话,很快它就包含了没有找到的文件。
我们同时还启动了大量的搜索线程。每个搜索线程从队列中取出一个文件,打开它,打印出包含关键字的所有行,然后取出下一个文件。我们使用了一个小技巧来在工作结束后终止线程。为了发出完成信号,枚举线程把一个虚拟对象放入队列。(这类似于在行李输送带上放一个写着“最后一个包”的虚拟包。)当搜索线程取到这个虚拟对象时,就将其放回并终止。
注意,这里不需要人任何显示的线程同步。在这个程序中,我们使用队列数据结构作为一种同步机制。
import java.io.*; import java.util.*; import java.util.concurrent.*; public class BlockingQueueTest { public static void main(String[] args) { Scanner in = new Scanner(System.in); System.out.print("Enter base directory (e.g. /usr/local/jdk1.6.0/src): "); String directory = in.nextLine(); System.out.print("Enter keyword (e.g. volatile): "); String keyword = in.nextLine(); final int FILE_QUEUE_SIZE = 10; final int SEARCH_THREADS = 100; BlockingQueue<File> queue = new ArrayBlockingQueue<File>(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(); } } /** * This task enumerates all files in a directory and its subdirectories. */ class FileEnumerationTask implements Runnable { /** * Constructs a FileEnumerationTask. * @param queue the blocking queue to which the enumerated files are added * @param startingDirectory the directory in which to start the enumeration */ public FileEnumerationTask(BlockingQueue<File> queue, File startingDirectory) { this.queue = queue; this.startingDirectory = startingDirectory; } public void run() { try { enumerate(startingDirectory); queue.put(DUMMY); } catch (InterruptedException e) { } } /** * Recursively enumerates all files in a given directory and its subdirectories * @param directory the directory in which to start */ public void enumerate(File directory) throws InterruptedException { File[] files = directory.listFiles(); for (File file : files) { if (file.isDirectory()) enumerate(file); else queue.put(file); } } public static File DUMMY = new File(""); private BlockingQueue<File> queue; private File startingDirectory; } /** * This task searches files for a given keyword. */ class SearchTask implements Runnable { /** * Constructs a SearchTask. * @param queue the queue from which to take files * @param keyword the keyword to look for */ public SearchTask(BlockingQueue<File> queue, String keyword) { this.queue = queue; this.keyword = keyword; } 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 (IOException e) { e.printStackTrace(); } catch (InterruptedException e) { } } /** * Searches a file for a given keyword and prints all matching lines. * @param file the file to search */ public void search(File file) throws IOException { Scanner in = new Scanner(new FileInputStream(file)); int lineNumber = 0; while (in.hasNextLine()) { lineNumber++; String line = in.nextLine().trim(); if (line.contains(keyword)) System.out.printf("%s:%d %s%n", file.getPath(), lineNumber, line); } in.close(); } private BlockingQueue<File> queue; private String keyword; }
相关文章推荐
- Java阻塞队列线程集控制的实现
- java多线程:16、阻塞队列【BlockingQueue】实现线程数据共享的效果
- 18-java5阻塞队列实现线程间通信-实现线程间通信方式(3)
- JAVA可阻塞队列-ArrayBlockingQueue子类BlockingQueue的应用,使用它来实现子线程打印10次,主线程打印100次,如此反复
- 黑马程序员——Java5中的线程并发库(二)--Semaphore实现信号灯、其他同步函数、可阻塞的队列、同步集合
- Java -- 使用阻塞队列(BlockingQueue)控制线程通信
- java使用阻塞队列(BlockingQueue)来控制线程通信
- Java中使用阻塞队列控制线程集实例
- Java阻塞队列BlockingQueue实现生产者消费者-只有代码-不讲原理
- Java线程超时控制的实现
- 线程高级应用-心得7-java5线程并发库中阻塞队列Condition的应用及案例分析
- Java线程:新特征-阻塞队列
- JAVA线程15 - 新特性:阻塞队列&阻塞栈
- Java阻塞队列(BlockingQueue)实现 生产者/消费者 示例
- 线程高级应用-心得7-java5线程并发库中阻塞队列Condition的应用及案例分析
- java线程 超时控制的实现
- JAVA与多线程开发(线程基础、继承Thread类来定义自己的线程、实现Runnable接口来解决单继承局限性、控制多线程程并发)
- Java阻塞队列的实现
- (13)多线程与并发库之java5阻塞队列(BlockingQueue)的应用----子线程循环10次,接着主线程循环100次,接着又回到子线程循环10次,接着再回到主线程循环100次,如此循环50次
- java实现线程池,实现线程队列(转)