Java ThreadPoolExecutor

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这样的东西。然后，你可以使用像芭乐的Iterables.concat这样的东西加入它们，只有一个与Fastutil的LongList兼容的列表版本。

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