JavaExecutorServiceinvokeAll多任務決議

我正在使用執行程式服務并行運行任務。并行運行方法接受輸入 integer 并回傳 integer 。由于并行任務具有回傳型別，所以我使用了 Callable 匿名類。您可以在下面的示例中看到ExecutorServiceExample task(int i )從 executer 呼叫。任務方法也有 1 秒的等待時間并拋出例外i==7;

在下面的實作中，我使用 invokeAll 并使用 isDone 并嘗試收集資料。

下面的程式拋出IllegalMonitorStateException.

Future 任務迭代和檢查 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();
        }
    }
}

uj5u.com熱心網友回復：

事實上，每個任務都會產生一個IllegalMonitorStateException因為你沒有wait在synchronized塊中呼叫方法：IllegalMonitorStateException on wait() call。也許你應該使用sleep而不是wait.

ExecutionException被 拋出future#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();
                }
            }

uj5u.com熱心網友回復：

我不知道你為什么設計成這樣。但顯然，它有很多問題。 i/7==0 or i % 7 ==0? examples不是鎖，那么為什么要使用等待呢？必須完成“invokeAll”回傳的期貨，但在呼叫 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();
            }
        }


    }
}

標籤：

上一篇：python中的有序semaphore.aquire()

下一篇：fork()執行緒：分治陣列的總和