Java 线程池执行器

Question

我在理解 Java ThreadPoolExecutor 时遇到了很大的麻烦。例如，我想计算数字 1-1000 的平方：

public static void main(String[] args) throws InterruptedException, ExecutionException {

    Callable<ArrayList<Integer>> c = new squareCalculator(1000);
    ExecutorService executor = Executors.newFixedThreadPool(5);
    Future<ArrayList<Integer>> result = executor.submit(c);


    for(Integer i: result.get()){
        System.out.println(i);
    }

}

还有

public class squareCalculator implements Callable<ArrayList<Integer>>{
    private int i;
    private int max;

    private int threadID;
    private static int id;
    private ArrayList<Integer> squares;

    public squareCalculator(int max){
        this.max = max;
        this.i = 1;
        this.threadID = id;
        id++;
        squares = new ArrayList<Integer>();

    }

    public ArrayList<Integer> call() throws Exception {
        while(i <= max){            
            squares.add(i*i);
            System.out.println("Proccessed number " +i + " in thread "+this.threadID);
            Thread.sleep(1);
            i++;
        }
        return squares;

    }
}

现在我的问题是，我只有一个线程进行计算。我希望得到 5 个线程。

Answer 1

如果你想让Callable同时运行5次，你需要submit它5次。 你只提交了一次，然后问了5次结果。

submit()的Javadoc:

提交一个返回值的任务执行并返回一个 Future 表示任务的未决结果。这 Future 的 get 方法将返回任务的结果 成功完成。

您会看到 submit() 的 Javadoc 使用单数表示“任务”，而不是“任务”。

修复很简单：多次提交：

Future<ArrayList<Integer>> result1 = executor.submit(c);
Future<ArrayList<Integer>> result2 = executor.submit(c);
Future<ArrayList<Integer>> result3 = executor.submit(c);
/// etc..

result1.get();
result2.get();
result3.get();
// etc..

Answer 2

ExecutorService 将使用一个线程来执行您提交的每个Callable 任务。因此，如果你想让多个线程计算平方，你必须提交多个任务，例如每个数字一个任务。然后，您会从每个任务中获得一个 Future<Integer>，您可以将其存储在一个列表中并在每个任务上调用 get() 以获取结果。

public class SquareCalculator implements Callable<Integer> {
    private final int i;

    public SquareCalculator(int i) {
        this.i = i;
    }

    @Override
    public Integer call() throws Exception {
        System.out.println("Processing number " + i + " in thread " + Thread.currentThread().getName());
        return i * i;
    }

    public static void main(String[] args) throws Exception {
        ExecutorService executor = Executors.newFixedThreadPool(5);
        List<Future<Integer>> futures = new ArrayList<>();

        // Create a Callable for each number, submit it to the ExecutorService and store the Future
        for (int i = 1; i <= 1000; i++) {
            Callable<Integer> c = new SquareCalculator(i);
            Future<Integer> future = executor.submit(c);
            futures.add(future);
        }

        // Wait for the result of each Future
        for (Future<Integer> future : futures) {
            System.out.println(future.get());
        }

        executor.shutdown();
    }
}

然后输出看起来像这样：

Processing number 2 in thread pool-1-thread-2
Processing number 1 in thread pool-1-thread-1
Processing number 6 in thread pool-1-thread-1
Processing number 7 in thread pool-1-thread-2
Processing number 8 in thread pool-1-thread-2
Processing number 9 in thread pool-1-thread-2
...
1
4
9
...

Answer 3

这是一个尝试并行执行的有趣问题，因为创建结果数组（或列表）在 O(n) 时间内运行，因为它在创建时被初始化为零。

public static void main(String[] args) throws InterruptedException {
    final int chunks = Runtime.getRuntime().availableProcessors();
    final int max = 1001;

    ExecutorService executor = Executors.newFixedThreadPool(chunks);

    final List<ArrayList<Long>> results = new ArrayList<>(chunks);

    for (int i = 0; i < chunks; i++) {
        final int start = i * max / chunks;
        final int end = (i + 1) * max / chunks;
        final ArrayList<Long> localResults = new ArrayList<>(0);
        results.add(localResults);

        executor.submit(new Runnable() {
            @Override
            public void run() {
                // Reallocate enough space locally so it's done in parallel.
                localResults.ensureCapacity(end - start);
                for (int j = start; j < end; j++) {
                    localResults.add((long)j * (long)j);
                }
            }
        });
    }

    executor.shutdown();
    executor.awaitTermination(Long.MAX_VALUE, TimeUnit.MICROSECONDS);

    int i = 0;
    for (List<Long> list : results) {
        for (Long l : list) {
            System.out.printf("%d: %d\n", i, l);
            ++i;
        }
    }

}

在这里，处理包装类的开销会降低性能，因此您应该使用 Fastutil 之类的东西。然后，您可以使用 Guava 的 Iterables.concat 之类的东西加入它们，这只是一个与 Fastutil 的 LongList 兼容的 List 版本。

这也可能是一个很好的 ForkJoinTask，但同样，您需要高效的逻辑（映射，而不是复制；与 List.sublist 相反）列表连接函数来实现加速。