快捷導(dǎo)航

Java中使用阻塞隊(duì)列控制線程集實(shí)例

更新時間：2015年01月15日 11:08:18 投稿：junjie

這篇文章主要介紹了Java控制阻塞隊(duì)列線程集實(shí)例,本文用一個程序展示了如何使用阻塞隊(duì)列來控制線程集,程序功能是在一個目錄及它的所有子目錄下搜索所有文件,打印出包含指定關(guān)鍵字的文件列表,需要的朋友可以參考下

隊(duì)列以一種先進(jìn)先出的方式管理數(shù)據(jù)。如果你試圖向一個已經(jīng)滿了的阻塞隊(duì)列中添加一個元素，或是從一個空的阻塞隊(duì)列中移除一個元素，將導(dǎo)致線程阻塞。在多線程進(jìn)行合作時，阻塞隊(duì)列是很有用的工具。工作者線程可以定期的把中間結(jié)果存到阻塞隊(duì)列中。而其他工作者線程把中間結(jié)果取出并在將來修改它們。隊(duì)列會自動平衡負(fù)載。如果第一個線程集運(yùn)行的比第二個慢，則第二個線程集在等待結(jié)果時就會阻塞。如果第一個線程集運(yùn)行的快，那么它將等待第二個線程集趕上來。

下面的程序展示了如何使用阻塞隊(duì)列來控制線程集。程序在一個目錄及它的所有子目錄下搜索所有文件，打印出包含指定關(guān)鍵字的文件列表。

java.util.concurrent包提供了阻塞隊(duì)列的4個變種：LinkedBlockingQueue、ArrayBlockingQueue、PriorityBlockingQueue和DelayQueue。我們用的是ArrayBlockingQueue。ArrayBlockingQueue在構(gòu)造時需要給定容量，并可以選擇是否需要公平性。如果公平參數(shù)被設(shè)置了，等待時間最長的線程會優(yōu)先得到處理。通常，公平性會使你在性能上付出代價，只有在的確非常需要的時候再使用它。

生產(chǎn)者線程枚舉在所有子目錄下的所有文件并把它們放到一個阻塞隊(duì)列中。這個操作很快，如果隊(duì)列沒有設(shè)上限的話，很快它就包含了沒有找到的文件。

我們同時還啟動了大量的搜索線程。每個搜索線程從隊(duì)列中取出一個文件，打開它，打印出包含關(guān)鍵字的所有行，然后取出下一個文件。我們使用了一個小技巧來在工作結(jié)束后終止線程。為了發(fā)出完成信號，枚舉線程把一個虛擬對象放入隊(duì)列。（這類似于在行李輸送帶上放一個寫著“最后一個包”的虛擬包。）當(dāng)搜索線程取到這個虛擬對象時，就將其放回并終止。

注意，這里不需要人任何顯示的線程同步。在這個程序中，我們使用隊(duì)列數(shù)據(jù)結(jié)構(gòu)作為一種同步機(jī)制。

復(fù)制代碼代碼如下:

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

您可能感興趣的文章: