JavaExecutorService调用所有多个任务解析

Question

我正在使用执行器服务并行运行任务。并行运行方法采用输入整数并返回整数 。由于并行任务具有返回类型，因此我使用了可调用的匿名类。您可以在下面的示例中看到 ExecutorServiceExample task（int i ） 是从 executer 调用的。任务方法也有1秒的等待时间，并为i==7抛出异常;

在下面的实现中，我使用invokeAll和isDone，并尝试收集数据。

下面的程序抛出IllegalMonitorStateException。

未来任务迭代和检查有什么问题 isDone 和 get（） 。如何处理特定调用的异常。我想并行运行所有1到14个任务，并在所有完成时收集返回类型。另外，在错误的情况下，要知道输入，它给出了异常（7和14）

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.*;
import java.util.stream.Collectors;

class MyException extends Exception{
    MyException(String message) {
        super(message);
    }
}

public class ExecutorServiceExample {

    public int task(int i) throws MyException, InterruptedException {
        System.out.println("Running task.."+i);
        wait(1000);
        if(i%7==0) {
            throw new MyException("multiple of 7 not allowed");
        }

        return i;
    }

    public static void main(String[] args) {

        ExecutorService executorService = Executors.newFixedThreadPool(10);

        List<Callable<Integer>> tasks = Arrays.asList(1,2,3,4,5,6,7,8,9,10,11,12,13,14).stream().map(id->{
           return new Callable<Integer>() {
               @Override
               public Integer call() throws Exception {
                   ExecutorServiceExample executorServiceExample = new ExecutorServiceExample();
                   return executorServiceExample.task(id);
               }
           };
        }).collect(Collectors.toList());
        
        try{
            List<Future<Integer>> results = executorService.invokeAll(tasks);

            for (Future<Integer> task: results) {
                if(task.isDone()){
                    System.out.println(task.get());
                }
            }

        }catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }finally {
            executorService.shutdown();
        }
    }
}

Answer 1

我不知道你为什么要这样设计它。但是很明显，它有很多问题。i/7==0或i%7==0？示例不是锁，所以为什么要使用等待？“调用所有”返回的期货必须完成，但可能会在调用get时获得异常。这是你想要的吗？

import java.util.List;
import java.util.concurrent.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;

class MyException extends Exception {
    MyException(String message) {
        super(message);
    }
}

public class ExecutorServiceExample {

    public int task(int i) throws MyException, InterruptedException {
        TimeUnit.MILLISECONDS.sleep(1000);
        if (i % 7 == 0) {
            throw new MyException("multiple of 7 not allowed");
        }

        return i;
    }

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

        ExecutorService executorService = Executors.newFixedThreadPool(10);

        List<Callable<Integer>> tasks = Stream.of(1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14)
                .map(id -> (Callable<Integer>) () -> {
                    ExecutorServiceExample executorServiceExample = new ExecutorServiceExample();
                    return executorServiceExample.task(id);
                }).collect(Collectors.toList());
        List<Future<Integer>> results = executorService.invokeAll(tasks);
        executorService.shutdown();
        for (Future<Integer> task : results) {
            try {
                System.out.println(task.get());
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        }


    }
}

Answer 2

事实上，每个任务都会产生一个＜code＞IllegalMonitorStateException＜/code＞因为在＜code＞synchronized＜/code＞块中没有调用＜code＞wait＜/code＞方法：Illegal MonitorStateExeption on wait（）调用。也许你应该使用睡眠，而不是等待。

ExecutionException由未来#get抛出。因此，如果您缩小try-catch的范围，实际上将捕获14个异常：

           for (Future<Integer> task: results) {
                try {
                    System.out.println(task.get());
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } catch (ExecutionException e) {
                    e.printStackTrace();
                }
            }