java - LinkedBlockingQueue 无法等待线程执行

Question

我正在制作一个WordCounter，它有几个线程计算不同文件中的单词。除了一个小问题，我已经把一切都搞定了。

我无法找出等待线程完成的正确方法。如果我设置 Thread.sleep 等待一小段时间，一切都会正常，唯一的问题是，如果计数器花费的时间比 sleep 时间长，这将不是一个正确的解决方案。

import java.io.*;
import java.util.*;
import java.util.concurrent.BlockingQueue;

public class WordCounter implements Runnable{
    private String file;
    private BlockingQueue<Integer> q;
    private int words = 0;

    public WordCounter(String f, BlockingQueue<Integer> queue){
        file = f;
        q = queue;
    }

    public void run(){
        try{
            Scanner in = new Scanner(new File(file));

            while (in.hasNext()){
                in.next();
                words++;
            }
            in.close();
            System.out.println(file + ": " + words);
            q.add(words);
        }

        catch (FileNotFoundException e){
            System.out.println(file + " blev ikke fundet.");
        }
    }

}

这是来自实际字计数器的代码。我希望我的主线程等待这些单词计数器线程执行 q.add(words);在做其他事情之前先运行函数。

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

public class MainThread implements Runnable{

    private String[] arguments;

    public MainThread(String[] args){
        arguments = args;
    }

    public void run() {

        final BlockingQueue<Integer> queue = new LinkedBlockingQueue<Integer>();

        for(String f : arguments){
            WordCounter c = new WordCounter(f, queue);
            Thread t = new Thread(c);
            t.start();  
        }
        while(!queue.isEmpty()){
            try {
                System.out.println(queue.take());
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

这是主线程。在继续底部的 while 语句之前，我需要某种方法来等待其他线程完成，但是如何实现呢？

提前致谢。

Answer 1

使用 ExecutorService 并等待 Future 返回。下面的代码将把每个任务提交给执行器服务(线程池)中的一个线程，并获取该任务的 future 。当所有提交后，它将等待 future 。 get方法只有在run时才会返回方法在任务中完成。

public class MainThread implements Runnable{

    private String[] arguments;

    public MainThread(String[] args){
        arguments = args;
    }

    public void run() {

        ExecutorService e = Executors.newFixedThreadPool(arguments.length);
        final BlockingQueue<Integer> queue = new LinkedBlockingQueue<Integer>();
        List<Future<?>> tasks = new ArrayList<>();
        for(String f : arguments){
            tasks.add(e.submit(new WordCounter(f, queue)));
        }
        for(Future<?> f : tasks) {
            f.get();
        }
        while(!queue.isEmpty()){
            try {
                System.out.println(queue.take());
                Thread.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

但是

您可以通过完全删除 BlockingQueue 并将每个任务设为 Callable<Integer> 来使代码更简洁。它只是返回 words多变的。当您调用future.get()时返回值是字数。