Java。带阻塞队列的消费者生产者。搜索工具

Question

我试图用阻塞队列实现一些消费者-生产者问题。为了达到某种目的，我决定编写文件搜索工具。

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

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

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

下面是一些代码：

搜索机制：

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

如你所说，我试着这样做:

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

搜索：

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，因为我需要一些机制来关闭主类中的线程池。正如你所说，它也使用pool（超时），我的线程终于完成了他们的工作。

检查中

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个线程将作为ContainingCheck算法运行？

搜索算法将搜索文件超过 1 秒，并且包含检查算法完成工作？显然，我可以将超时更改为 2 秒甚至更多，但我们总是试图优化我们的程序。

Answer 2

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

而且您不需要为每个搜索算法创建新的线程池。