Item 80: Prefer executors, tasks, and streams to threads

80.スレッドよりエグゼキュータ、タスク、ストリームを選択すべし

• Executor Frameworkを用いて簡単にバックグラウンドで処理を実行させることができる。
• Executor Frameworkを用いることで、全てのタスクが終了する、またはどれか一つのタスクが終了するまで待つことや、周期的にタスクを実行するなどといったことができるようになった。

package tryAny.effectiveJava;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class ExecutorTest {
    public static void main(String[] args) throws InterruptedException, ExecutionException {
        List<Callable<String>> tasks = Arrays.asList(() -> "first", () -> "second");

        ExecutorService es = Executors.newFixedThreadPool(2);
        List<Future<String>> rets = es.invokeAll(tasks);

        for (Future<String> r : rets) {
            System.out.println(r.get());
        }

        System.out.println(es.invokeAny(tasks));// first か second ランダムで

    }
}

package tryAny.concurrentUtility;

import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

// １秒ごとにランダム値を生成し続ける。
public class ConcurrentTest6 {
    public static void main(String[] args) {
	ScheduledThreadPoolExecutor stpe = new ScheduledThreadPoolExecutor(1);

	ScheduledFuture<?> sf = stpe.scheduleAtFixedRate(new Runnable() {
	    @Override
	    public void run() {
		System.out.println(Math.random());
	    }
	}, 1000, 1000, TimeUnit.MILLISECONDS);

	while (true) {
	    if (sf.isCancelled() || sf.isDone()) {
		stpe.shutdown();
		break;
	    }
	}
    }
}

• 複数のスレッドを利用した処理がしたい場合には、java.util.concurrent.Executors に様々なスレッドプールを作成するファクトリメソッドがあるので、それを使う。それらの使用では用途に合わない場合、ThreadPoolExecutorを直接使用する。
• 比較的処理の軽いサーバにおいては、newCachedThreadPool を使いうるが、cached thread pool はサブミットされたタスクをすぐに実行に移し、スレッドがなければ新しいスレッドを作る、という挙動を取るため処理が重いサーバでは newFixedThreadPool を使うべき。
• Threadsを直接使用するのは避けるべき。ExecutorFramework　では、処理の単位と実行のメカニズムが分離されており、柔軟性がある。
• Java7以降、fork-join という仕組みで効率的に処理を実行できるようになった。

package tryAny.concurrentUtility;

import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.RecursiveAction;

public class ConcurrentTest8 {
    public static void main(String[] args) {
        int data[] = { 1, 2, 3, 4, 5, 6, 7 };
        ForkJoinPool service = new ForkJoinPool();
        service.invoke(new AddAction(data, 0, data.length));
    }
}

class AddAction extends RecursiveAction {
    private static final int THRESHOLD_SIZE = 3;
    private int start;
    private int end;
    private int[] numbers;

    public AddAction(int[] numbers, int start, int end) {
        this.start = start;
        this.end = end;
        this.numbers = numbers;
    }

    protected void compute() {
        int total = 0;
        if (end - start <= THRESHOLD_SIZE) {
            for (int i = start; i < end; i++) {
                total += numbers[i];
            }
            System.out.println(total + " ");
        } else {
            new AddAction(numbers, start + THRESHOLD_SIZE, end).fork();
            new AddAction(numbers, start, Math.min(end, start + THRESHOLD_SIZE)).compute();
        }
    }
}