Java中怎么利用阻塞隊列實(shí)現搜索

這期內容當中小編將會(huì )給大家帶來(lái)有關(guān)Java中怎么利用阻塞隊列實(shí)現搜索，文章內容豐富且以專(zhuān)業(yè)的角度為大家分析和敘述，閱讀完這篇文章希望大家可以有所收獲。

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

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

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

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

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

注意，這里不需要人任何顯示的線(xiàn)程同步。在這個(gè)程序中，我們使用隊列數據結構作為一種同步機制。

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