Java中使用阻塞队列控制线程集实例

所属分类: 软件编程 / java 阅读数: 56
收藏 0 赞 0 分享

队列以一种先进先出的方式管理数据。如果你试图向一个已经满了的阻塞队列中添加一个元素,或是从一个空的阻塞队列中移除一个元素,将导致线程阻塞。在多线程进行合作时,阻塞队列是很有用的工具。工作者线程可以定期的把中间结果存到阻塞队列中。而其他工作者线程把中间结果取出并在将来修改它们。队列会自动平衡负载。如果第一个线程集运行的比第二个慢,则第二个线程集在等待结果时就会阻塞。如果第一个线程集运行的快,那么它将等待第二个线程集赶上来。

下面的程序展示了如何使用阻塞队列来控制线程集。程序在一个目录及它的所有子目录下搜索所有文件,打印出包含指定关键字的文件列表。

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; 
}

更多精彩内容其他人还在看

SpringBoot中使用Ehcache的详细教程

EhCache 是一个纯 Java 的进程内缓存框架,具有快速、精干等特点,是 Hibernate 中默认的 CacheProvider。这篇文章主要介绍了SpringBoot中使用Ehcache的相关知识,需要的朋友可以参考下
收藏 0 赞 0 分享

在idea 中添加和删除模块Module操作

这篇文章主要介绍了在idea 中添加和删除模块Module操作,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧
收藏 0 赞 0 分享

java spring整合junit操作(有详细的分析过程)

这篇文章主要介绍了java spring整合junit操作(有详细的分析过程),具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧
收藏 0 赞 0 分享

详解JAVA 弱引用

这篇文章主要介绍了 JAVA 弱引用的相关资料,帮助大家更好的理解和学习java引用对象,感兴趣的朋友可以了解下
收藏 0 赞 0 分享

深入了解JAVA 虚引用

这篇文章主要介绍了JAVA 虚引用的相关资料,帮助大家更好的理解和学习JAVA,感兴趣的朋友可以了解下
收藏 0 赞 0 分享

详解JAVA 强引用

这篇文章主要介绍了JAVA 强引用的相关资料,帮助大家更好的理解和学习,感兴趣的朋友可以了解下
收藏 0 赞 0 分享

java中的按位与(&)用法说明

这篇文章主要介绍了java中的按位与(&)用法说明,具有很好的参考价值,希望对大家有所帮助。一起跟随小编过来看看吧
收藏 0 赞 0 分享

深入了解JAVA 软引用

这篇文章主要介绍了JAVA 软引用的相关资料,帮助大家更好的理解和学习,感兴趣的朋友可以了解下
收藏 0 赞 0 分享

利用MyBatis实现条件查询的方法汇总

这篇文章主要给大家介绍了关于利用MyBatis实现条件查询的相关资料,文中通过示例代码介绍的非常详细,对大家的学习或者使用MyBatis具有一定的参考学习价值,需要的朋友们下面来一起学习学习吧
收藏 0 赞 0 分享

Intellij IDEA 与maven 版本不符 Unable to import maven project See logs for details: No implementation for org.apache.maven.model.path.PathTranslator was bound

这篇文章主要介绍了Intellij IDEA 与maven 版本不符 Unable to import maven project See logs for details: No implementation for org.apache.maven.model.path.Pa
收藏 0 赞 0 分享
查看更多