java.util.concurrent.ScheduledExecutorService

何かそれなりに重い処理が色々あったとする。バッチ的なアレとか。
で、処理は大体がアトミックで、スレッドプールを使って、それぞれを処理する事で、処理効率を稼ぎたい時に使うAPI。

Javadocを見てても良く分からんくて、ハマったので、実装例をばメモ。

このコードの内容は、

処理は重くていつ終わるか分からない上に、処理中に更に新しいタスクを積んでしまうので、
全体として、本当にいつ終わるか分からないので、
プロセスを起動した後、一定の時間が経過したら、それ以上タスクを積めない様にしつつ、
プロセスを綺麗に終了する。

という感じ。

import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class Heavy {
  protected ScheduledExecutorService executorService;

  public Heavy() {
  }

  public void initialize() {
    this.executorService = Executors.newScheduledThreadPool(5);
    this.executorService.schedule(new Runnable() {
      @Override
      public void run() {
        executorService.shutdown();
      }
    }, 3, TimeUnit.MINUTES); // 3分後にシャットダウン開始。
  }

  public void execute() {
    this.executorService.schedule(new Runnable() {
      @Override
      public void run() {
        // スゲェ重い処理。
        execute(); // 再処理
      }
    }, 0, TimeUnit.MILLISECONDS);
  }

  public void dispose() {
    try {
      this.executorService.awaitTermination(4, TimeUnit.MINUTES);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
  }

  public static void main(String[] args) {
    Heavy me = new Heavy();
    try {
      me.initialize();

      // 処理をキューイングする。
      me.execute();
      me.execute();
      me.execute();
      me.execute();
    } finally {
      me.dispose();
    }
  }
}

ポイントは、

• ExecutorService#shutdownを呼ぶRunnableを最初にscheduleしておく事
• shutdownキューが実行されるであろうより後に、実行されるようにExecutorService#awaitTerminationを呼ぶ事。
  • awaitTerminationは、設定した時間が経過するまでは、処理がブロックさりる事が重要。