並行処理いろいろ

Runnableを使用した場合

//Runnableオブジェクトを生成する
Runnable r = () -> {
    System.out.println("Task is executed.");
    System.out.println("end");
};

//Executorを使用して実行
ExecutorService executor = Executors.newSingleThreadExecutor();
executor.execute(r);

//Executorは通常ファクトリメソッドを使用して生成する
Executors.newSingleThreadExecutor();
Executors.newCachedThreadPool();
Executors.newFixedThreadPool(3);
Executors.newScheduledThreadPool(2);
//違いはよくわからない(勉強中)

Callableを使用する場合

//Runnableでは戻り値や例外処理ができないため、
//そのような場合はCallbele<T>を使用する
Callable<String> call = () -> {
    return "called!";
};
//Callableを使用する場合はExecutorのsubmitメソッドを使用する
//futureオブジェクトが返却される
Future<String> future = executor.submit(call);
//Callableの実装の戻り値を取得する場合は#get()を使用する
System.out.println(future.get());

なお、タスクの実行が終了しても、Executorそのものは自動的に終了しない。
そのため、#shutdown()を明示的に呼び出す必要がある。

//Executorの終了
executor.shutdown();

ちなみに、Executorの終了後にタスクを実行すると例外が発生する

executor.shutdown();
//Executorの終了後にタスクを実行すると、
//java.util.concurrent.RejectedExecutionExceptionが発生
executor.submit(call);

Fork/Join フレームワーク

//RecursiveActionを継承したクラスを作る
public class MyAction extends RecursiveAction {

    @Override
    protected void compute() {
        
        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        System.out.println("MyAction!");
    }

}

//実行側
    public static void main(String[] args) throws Exception {
        ForkJoinPool executor = new ForkJoinPool();
        executor.execute(new MyAction());
        System.out.println("main:ok");
        Thread.sleep(3000);
        System.out.println("done!");
    }

Streamを使った並列処理

List<Integer> list = new ArrayList<>(Arrays.asList(1,2,3,4,5));
//parallelStreamを使用してStream化した場合、並列ストリームになる
//以下のコードは2乗した数字が出る順序は不定
list.parallelStream().map(i -> i * i).forEach(System.out::println);

//順次ストリームへの変換
list.parallelStream()
.map(i -> i * i)
.sequential() //このsequential()を呼び出すことで順次制御になる
.forEach(System.out::println);
//値は一定になる