爪哇。消费者 - 使用 BlockingQueue 的生产者。搜索工具

Question

我试图用 BlockingQueue 实现一些消费者-生产者问题。出于某种目的，我决定编写文件搜索工具。

我决定搜索机制是递归工作的，每个新目录都会有新的线程池来提高搜索速度。

我的问题是，我不知道如何实现最终停止打印线程（消费者）的机制 - 当搜索线程完成工作时。

我试图用像毒丸这样的想法来做到这一点，但效果不佳（线程在打印任何结果之前停止）。有什么想法我该怎么做？

这里有一些代码：

搜索机制：

public class SearchingAlgorithm implements Runnable {

private final File file;
private BlockingQueue<File> queue;
private ExecutorService executor;

public SearchingAlgorithm(File fileName, BlockingQueue<File> queue) {
    this.file = fileName;
    this.queue = queue;
    this.executor = Executors.newWorkStealingPool();
}

@Override
public void run() {
    try {
        searchDeep();
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}

private void searchDeep() throws InterruptedException {
    File[] files = file.listFiles();
    if (files != null) {
        for (File fil : files) {
            if (fil.isDirectory()) {
                executor.submit(new SearchingAlgorithm(fil, this.queue));
            } else {
                this.queue.add(fil);
            }
        }
    }
}

}

打印机：

public class ContainingCheckAlgorithm implements Runnable {

private BlockingQueue<File> queue;
// private ExecutorService executor;
private String keyWord;

public ContainingCheckAlgorithm(BlockingQueue<File> queue, String keyWord) {
    this.queue = queue;
    this.keyWord = keyWord;
    // executor = Executors.newFixedThreadPool(2);
}

@Override
public void run() {
    try {
        printFile();
    } catch (InterruptedException e) {
        e.printStackTrace();
    }

}

private void printFile() throws InterruptedException {
    while (true) {
        File takeFile = queue.take();
        String fileName = takeFile.getAbsolutePath()
                .toLowerCase();
        boolean isContainingKeyWord = fileName.contains(keyWord.toLowerCase());

        if (isContainingKeyWord) {
            System.out.println(takeFile.getAbsolutePath());
        }
    }
}

}

主要测试类：

public class MainClass {

public static void main(String[] args) throws InterruptedException {
    ExecutorService executor = Executors.newFixedThreadPool(2);
    BlockingQueue<File> queue = new LinkedBlockingQueue<>();

    File fileName = new File("C:/");

    SearchingAlgorithm sa = new SearchingAlgorithm(fileName, queue);
    executor.submit(sa);

    ContainingCheckAlgorithm ca = new ContainingCheckAlgorithm(queue, "Slipknot");
    executor.submit(ca);

    executor.shutdown();
}

}

Answer 1

将整个工作分为两个阶段。在第一阶段，如果队列为空，SearchingAlgorithm 的工作和 ContainingCheckAlgorithm 会等待新的工作。在第二阶段，所有 SearchingAlgorithm 实例都完成了，如果发现队列为空，则 ContainingCheckAlgorithm 退出。为了检测队列是否为空，ContainingCheckAlgorithm 使用 queue.poll(timeout) 而不是 queue.take()。

而且您不需要为每个 SearchingAlgorithm 创建新的线程池。

Answer 2

正如你所说，我尝试这样做：

搜索算法与其他搜索算法实例共享线程池。

正在搜索：

public class SearchingAlgorithm implements Runnable {

private final File file;
private BlockingQueue<File> queue;
private ExecutorService executor;

public SearchingAlgorithm(File fileName, BlockingQueue<File> queue, ExecutorService executor) {
    this.file = fileName;
    this.queue = queue;
    this.executor = executor;
}

@Override
public void run() {
    try {
        searchDeep();
    } catch (InterruptedException e) {
        e.printStackTrace();
    }
}

private void searchDeep() throws InterruptedException {
    File[] files = file.listFiles();
    if (files != null) {
        for (File fil : files) {
            if (fil.isDirectory()) {
                executor.submit(new SearchingAlgorithm(fil, this.queue, executor));
            } else {
                this.queue.add(fil);
            }
        }
    }
}

现在 ContainingCheckAlgorith 需要与主类共享 CountDownLatch，因为我需要一些机制来关闭主类中的线程池。正如你所说，它还使用池（超时），我的线程终于完成了他们的工作。

检查

public class ContainingCheckAlgorithm implements Runnable {

private BlockingQueue<File> queue;
private String keyWord;
private CountDownLatch latch;

public ContainingCheckAlgorithm(BlockingQueue<File> queue, String keyWord, CountDownLatch latch) {
    this.queue = queue;
    this.keyWord = keyWord;
    this.latch = latch;
}

@Override
public void run() {
    try {
        printFile();
        latch.countDown();
    } catch (InterruptedException e) {
        e.printStackTrace();
    }

}

private void printFile() throws InterruptedException {
    File takeFile;
    while ((takeFile = queue.poll(1, TimeUnit.SECONDS)) != null) {
        String fileName = takeFile.getName()
                .toLowerCase();
        boolean isContainingKeyWord = fileName.contains(keyWord.toLowerCase());

        if (isContainingKeyWord) {
            System.out.println(takeFile.getAbsolutePath());
        }
    }
}

主要：

public class MainClass {

public static void main(String[] args) throws InterruptedException {
    ExecutorService executor = Executors.newCachedThreadPool();
    BlockingQueue<File> queue = new LinkedBlockingQueue<>();
    CountDownLatch latch = new CountDownLatch(1);

    File fileName = new File("C:/");

    SearchingAlgorithm sa = new SearchingAlgorithm(fileName, queue, executor);
    executor.submit(sa);

    ContainingCheckAlgorithm ca = new ContainingCheckAlgorithm(queue, "Slipknot", latch);
    executor.submit(ca);

    latch.await();
    executor.shutdown();
}

看起来很奇怪，但我想知道如果：

• 超过 1 个线程将作为 ContainingCheckAlgorithm 运行？

• SearchingAlgorithm 会搜索超过 1 秒的文件，并且 ContainingCheckAlgorithm 完成工作？显然，我可以将超时更改为 2 秒，甚至更多，但我们总是尝试优化我们的程序。