Java 執(zhí)行器

Java線程教程 - Java執(zhí)行器

框架提供了一種將任務(wù)提交與任務(wù)執(zhí)行分離的方法。

java.util.concurrent包中的Executor接口是執(zhí)行器框架的基礎(chǔ)。

它是一個只有一個方法的接口，如圖所示：

public interface  Executor  {
    void  execute  (Runnable command);
}

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

class RunnableTask implements Runnable {
  private int taskId;
  private int loopCounter;

  public RunnableTask(int taskId, int loopCounter) {
    this.taskId = taskId;
    this.loopCounter = loopCounter;
  }
  public void run() {
    for (int i = 1; i <= loopCounter; i++) {
      try {
        System.out.println("Task #" + this.taskId + "  - Iteration #" + i);
        Thread.sleep(1000);
      } catch (Exception e) {
        System.out.println("Task #" + this.taskId
            + "  has  been  interrupted.");
        break;
      }
    }
  }
}
public class Main {
  public static void main(String[] args) {
    final int THREAD_COUNT = 3;
    final int LOOP_COUNT = 3;
    final int TASK_COUNT = 5;

    // Get an executor with three threads in its thread pool
    ExecutorService exec = Executors.newFixedThreadPool(THREAD_COUNT);

    // Create five tasks and submit them to the executor
    for (int i = 1; i <= TASK_COUNT; i++) {
      RunnableTask task = new RunnableTask(i, LOOP_COUNT);
      exec.submit(task);
    }
    exec.shutdown();
  }
}

上面的代碼生成以下結(jié)果。

結(jié)果承載任務(wù)

要在任務(wù)完成時獲取任務(wù)的結(jié)果，請使用Callable接口的實例。

類型參數(shù)V是任務(wù)的結(jié)果的類型。

Callable接口有一個call()方法。它可以返回任何類型的值。

它允許你拋出異常。它聲明如下：

public interface  Callable<V>  {
    V  call() throws   Exception;
}

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

class CallableTask implements Callable<Integer> {
  private int taskId;

  public CallableTask(int taskId) {
    this.taskId = taskId;
  }

  public Integer call() throws InterruptedException {
    int total = taskId;
    try {
      System.out.println("Task #" + this.taskId);
      Thread.sleep(1000);
    } catch (InterruptedException e) {
      System.out.println("Task #" + this.taskId
          + "  has  been  interupted.");
      throw e;
    }
    total+=taskId;
    return total;
  }
}

public class Main {
  public static void main(String[] args) throws Exception {
    // Get an executor with three threads in its thread pool
    ExecutorService exec = Executors.newFixedThreadPool(3);
    CallableTask task = new CallableTask(1);
    // Submit the callable task to executor
    Future<Integer> submittedTask = exec.submit(task);

    Integer result = submittedTask.get();
    System.out.println("Task"s total  sleep time: " + result + "  seconds");
    exec.shutdown();
  }
}

上面的代碼生成以下結(jié)果。

調(diào)度任務(wù)

執(zhí)行器框架允許您計劃將來運行的任務(wù)。

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

class ScheduledTask implements Runnable {
  private int taskId;

  public ScheduledTask(int taskId) {
    this.taskId = taskId;
  }

  public void run() {
    LocalDateTime currentDateTime = LocalDateTime.now();
    System.out.println("Task #" + this.taskId + "  ran  at "
        + currentDateTime);
  }
}
public class Main {
  public static void main(String[] args) {
    // Get an executor with 3 threads
    ScheduledExecutorService sexec = Executors.newScheduledThreadPool(3);

    ScheduledTask task1 = new ScheduledTask(1);
    ScheduledTask task2 = new ScheduledTask(2);

    // Task #1 will run after 2 seconds
    sexec.schedule(task1, 2, TimeUnit.SECONDS);

    // Task #2 runs after 5 seconds delay and keep running every 10 seconds
    sexec.scheduleAtFixedRate(task2, 5, 10, TimeUnit.SECONDS);

    try {
      TimeUnit.SECONDS.sleep(60);
    } catch (InterruptedException e) {
      e.printStackTrace();
    }
    sexec.shutdown();
  }
}

上面的代碼生成以下結(jié)果。

在任務(wù)執(zhí)行中處理未捕獲的異常

執(zhí)行器框架在任務(wù)執(zhí)行期間處理任何未捕獲異常的事件。

如果使用Executor對象的execute()方法執(zhí)行Runnable任務(wù)，任何未捕獲的運行時異常將停止任務(wù)執(zhí)行，并且異常堆棧跟蹤將打印在控制臺上。

import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class Main {
  public static void main(String[] args) {
    Runnable badTask = () -> {
      throw new RuntimeException(
          "Throwing exception  from  task execution...");
    };

    ExecutorService exec = Executors.newSingleThreadExecutor();
    exec.execute(badTask);
    exec.shutdown();
  }
}

上面的代碼生成以下結(jié)果。

在Callable任務(wù)中處理異常

以下代碼顯示了如何在Callable任務(wù)中處理異常。

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 Main {
  public static void main(String[] args) {
    Callable<Object> badTask = () -> {
      throw new RuntimeException(
          "Throwing exception from task execution...");
    };
    ExecutorService exec = Executors.newSingleThreadExecutor();
    Future submittedTask = exec.submit(badTask);
    try {
      Object result = submittedTask.get();
    } catch (ExecutionException e) {
      System.out.println(e.getMessage());
      System.out.println(e.getCause().getMessage());
    } catch (InterruptedException e) {
      e.printStackTrace();
    }

    exec.shutdown();
  }
}

執(zhí)行完成服務(wù)

要將提交的任務(wù)的結(jié)果提供給執(zhí)行程序，請使用執(zhí)行程序的完成服務(wù)。

它由CompletionService接口的一個實例表示。

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorCompletionService;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

class MyResult {
  private int taskId;
  private int result;

  public MyResult(int taskId, int result) {
    this.taskId = taskId;
    this.result = result;
  }

  public int getTaskId() {
    return taskId;
  }

  public int getResult() {
    return result;
  }

  public String toString() {
    return "Task  Name: Task  #" + taskId + ", Task  Result:" + result
        + "  seconds";
  }
}

class SleepingTask implements Callable<MyResult> {
  private int taskId;
  private int loopCounter;
  public SleepingTask(int taskId, int loopCounter) {
    this.taskId = taskId;
    this.loopCounter = loopCounter;
  }

  public MyResult call() throws InterruptedException {
    int totalSleepTime = 0;
    for (int i = 1; i <= loopCounter; i++) {
      try {
        System.out.println("Task #" + this.taskId + "  - Iteration #"
            + i);
        Thread.sleep(1000);
        totalSleepTime = totalSleepTime + 1000;
      } catch (InterruptedException e) {
        System.out.println("Task #" + this.taskId
            + "  has  been  interupted.");
        throw e;
      }
    }
    return new MyResult(taskId, totalSleepTime);
  }
}

public class Main {
  public static void main(String[] args) throws Exception {
    // Get an executor with three threads in its thread pool
    ExecutorService exec = Executors.newFixedThreadPool(3);

    // Completed task returns an object of the TaskResult class
    ExecutorCompletionService<MyResult> completionService = new ExecutorCompletionService<>(
        exec);
    for (int i = 1; i <= 5; i++) {
      SleepingTask task = new SleepingTask(i, 3);
      completionService.submit(task);
    }
    for (int i = 1; i <= 5; i++) {
      Future<MyResult> completedTask = completionService.take();
      MyResult result = completedTask.get();
      System.out.println("Completed a  task - " + result);
    }
    exec.shutdown();
  }
}

上面的代碼生成以下結(jié)果。