Java里ExecutorService的特性和使用
2022-04-156.0k 阅读
Java 里 ExecutorService 的特性和使用
ExecutorService 简介
在 Java 的并发编程领域，ExecutorService 是一个至关重要的接口，它扩展自 Executor 接口，为管理和控制异步任务执行提供了丰富的功能。Executor 接口只是定义了一个简单的 execute(Runnable task) 方法，用于提交一个任务来执行。而 ExecutorService 在此基础上增加了生命周期管理以及任务提交和执行结果获取等功能。
ExecutorService 允许我们提交 Runnable 或 Callable 类型的任务。Runnable 接口的任务没有返回值，而 Callable 接口的任务可以返回一个结果，并且在执行过程中可以抛出异常。ExecutorService 会将这些任务分配到线程池中执行，线程池是一组预先创建的线程，它们可以复用，从而避免了频繁创建和销毁线程带来的开销。
ExecutorService 的特性
生命周期管理
ExecutorService 提供了方法来管理其生命周期，主要包括以下三种状态：
- 运行（Running）：这是
ExecutorService的初始状态，它可以接受新的任务并执行已提交的任务。 - 关闭（Shutdown）：调用
shutdown()方法后，ExecutorService进入此状态。在此状态下，它不再接受新的任务，但会继续执行已提交的任务，直到所有任务完成。 - 终止（Terminated）：当所有已提交的任务（包括那些在关闭时已提交但尚未开始执行的任务）都执行完毕后，
ExecutorService进入此状态。可以通过调用isTerminated()方法来检查ExecutorService是否已终止。
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class LifeCycleExample {
public static void main(String[] args) {
ExecutorService executorService = Executors.newFixedThreadPool(2);
// 运行状态
System.out.println("Is Running: " +!executorService.isShutdown());
executorService.shutdown();
// 关闭状态
System.out.println("Is Shutdown: " + executorService.isShutdown());
// 等待所有任务完成
while (!executorService.isTerminated()) {
// 可以在此处进行其他操作，或者简单等待
}
// 终止状态
System.out.println("Is Terminated: " + executorService.isTerminated());
}
}
任务提交与执行
- 提交
Runnable任务：submit(Runnable task)方法用于提交一个Runnable任务，它返回一个Future对象。虽然Runnable任务本身没有返回值，但通过Future对象可以检查任务是否完成以及取消任务。 - 提交
Callable任务：submit(Callable<T> task)方法用于提交一个Callable任务，返回的Future对象可以获取任务执行的结果。如果任务尚未完成，调用Future.get()方法会阻塞当前线程，直到任务完成并返回结果。
import java.util.concurrent.*;
public class TaskSubmissionExample {
public static void main(String[] args) throws ExecutionException, InterruptedException {
ExecutorService executorService = Executors.newSingleThreadExecutor();
// 提交 Runnable 任务
Future<?> runnableFuture = executorService.submit(() -> {
System.out.println("Runnable task is running");
});
// 检查任务是否完成
System.out.println("Runnable task is done: " + runnableFuture.isDone());
// 提交 Callable 任务
Future<Integer> callableFuture = executorService.submit(() -> {
System.out.println("Callable task is running");
return 42;
});
// 获取 Callable 任务的结果
System.out.println("Callable task result: " + callableFuture.get());
executorService.shutdown();
}
}
线程池管理
ExecutorService 通常与线程池一起使用，线程池的类型决定了其性能和资源使用特性。常见的线程池类型有：
- 固定大小线程池（FixedThreadPool）：
Executors.newFixedThreadPool(int nThreads)创建一个固定大小的线程池，线程池中的线程数量始终保持为nThreads。当有新任务提交时，如果线程池中有空闲线程，则立即执行任务；否则任务会被放入队列中等待，直到有线程空闲。 - 缓存线程池（CachedThreadPool）：
Executors.newCachedThreadPool()创建一个缓存线程池，线程池的大小会根据任务的数量动态调整。如果线程池中有空闲线程，会复用这些线程；如果没有空闲线程，则创建新的线程来执行任务。当线程空闲 60 秒后，会被回收。 - 单线程线程池（SingleThreadExecutor）：
Executors.newSingleThreadExecutor()创建一个单线程的线程池，它保证所有任务按照顺序在一个线程中执行，避免了多线程竞争带来的问题。
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class ThreadPoolExample {
public static void main(String[] args) {
// 固定大小线程池
ExecutorService fixedThreadPool = Executors.newFixedThreadPool(3);
for (int i = 0; i < 5; i++) {
fixedThreadPool.submit(() -> {
System.out.println(Thread.currentThread().getName() + " is running");
});
}
// 缓存线程池
ExecutorService cachedThreadPool = Executors.newCachedThreadPool();
for (int i = 0; i < 5; i++) {
cachedThreadPool.submit(() -> {
System.out.println(Thread.currentThread().getName() + " is running");
});
}
// 单线程线程池
ExecutorService singleThreadExecutor = Executors.newSingleThreadExecutor();
for (int i = 0; i < 5; i++) {
singleThreadExecutor.submit(() -> {
System.out.println(Thread.currentThread().getName() + " is running");
});
}
fixedThreadPool.shutdown();
cachedThreadPool.shutdown();
singleThreadExecutor.shutdown();
}
}
ExecutorService 的使用场景
并行计算任务
在需要进行大量计算的场景中，例如数据处理、科学计算等，可以将任务分解为多个子任务，通过 ExecutorService 提交到线程池中并行执行，从而提高计算效率。
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
public class ParallelComputingExample {
public static void main(String[] args) throws ExecutionException, InterruptedException {
ExecutorService executorService = Executors.newFixedThreadPool(4);
List<Future<Integer>> futures = new ArrayList<>();
for (int i = 0; i < 10; i++) {
int number = i;
Future<Integer> future = executorService.submit(() -> {
// 模拟计算任务
return number * number;
});
futures.add(future);
}
for (Future<Integer> future : futures) {
System.out.println("Result: " + future.get());
}
executorService.shutdown();
}
}
异步任务处理
对于一些不需要立即获取结果的任务，如日志记录、数据缓存更新等，可以通过 ExecutorService 提交为异步任务，让主线程继续执行其他操作，提高系统的响应速度。
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
public class AsynchronousTaskExample {
public static void main(String[] args) {
ExecutorService executorService = Executors.newSingleThreadExecutor();
executorService.submit(() -> {
// 模拟日志记录任务
System.out.println("Logging task is running");
});
// 主线程继续执行其他操作
System.out.println("Main thread is doing other things");
executorService.shutdown();
}
}
任务调度与定时执行
ScheduledExecutorService 是 ExecutorService 的子接口，它提供了任务调度和定时执行的功能。可以使用 schedule() 方法在指定延迟后执行任务，或者使用 scheduleAtFixedRate() 或 scheduleWithFixedDelay() 方法周期性地执行任务。
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
public class TaskSchedulingExample {
public static void main(String[] args) {
ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(1);
// 在 2 秒后执行任务
scheduledExecutorService.schedule(() -> {
System.out.println("Delayed task is running");
}, 2, TimeUnit.SECONDS);
// 每 3 秒执行一次任务
scheduledExecutorService.scheduleAtFixedRate(() -> {
System.out.println("Periodic task is running");
}, 0, 3, TimeUnit.SECONDS);
// 为了演示，这里主线程睡眠 10 秒后关闭调度器
try {
Thread.sleep(10000);
} catch (InterruptedException e) {
e.printStackTrace();
}
scheduledExecutorService.shutdown();
}
}
深入理解 ExecutorService 的实现原理
线程池的工作流程
当使用 ExecutorService 提交任务时，线程池的工作流程如下：
- 任务提交：调用
submit()或execute()方法提交任务。 - 线程分配：如果线程池中有空闲线程，任务会立即分配给空闲线程执行。如果没有空闲线程，根据线程池的类型和队列的情况进行处理。
- 固定大小线程池：任务会被放入任务队列中等待，直到有线程空闲。
- 缓存线程池：如果任务队列已满，会创建新的线程来执行任务，除非线程数量达到最大限制。
- 单线程线程池：任务会被放入任务队列中，按顺序由唯一的线程执行。
- 任务执行：线程从任务队列中取出任务并执行。执行完毕后，线程回到空闲状态，等待执行下一个任务，或者根据线程池的配置被回收。
任务队列的作用
任务队列在 ExecutorService 中起着缓冲任务的作用。当线程池中的线程都在忙碌时，新提交的任务会被放入任务队列。常见的任务队列类型有：
- 无界队列（如
LinkedBlockingQueue）：这种队列可以容纳无限数量的任务。使用无界队列时，线程池中的线程数量通常不会超过核心线程数，因为新任务总是可以放入队列中等待执行。 - 有界队列（如
ArrayBlockingQueue）：有界队列有固定的容量。当队列满时，新任务的处理方式取决于线程池的饱和策略。
饱和策略
当任务队列已满且线程池中的线程数量达到最大限制时，ExecutorService 会采用饱和策略来处理新提交的任务。常见的饱和策略有：
- AbortPolicy（默认策略）：直接抛出
RejectedExecutionException异常，拒绝新任务的提交。 - CallerRunsPolicy：将任务交给调用者线程来执行，这样可以减轻线程池的压力，但会影响调用者线程的正常执行。
- DiscardPolicy：直接丢弃新提交的任务，不做任何处理。
- DiscardOldestPolicy：丢弃任务队列中最老的任务，然后尝试将新任务放入队列。
import java.util.concurrent.*;
public class RejectionPolicyExample {
public static void main(String[] args) {
ThreadPoolExecutor executor = new ThreadPoolExecutor(
2, // 核心线程数
4, // 最大线程数
1, // 线程存活时间
TimeUnit.MINUTES,
new ArrayBlockingQueue<>(2), // 有界队列
new ThreadPoolExecutor.CallerRunsPolicy() // 饱和策略
);
for (int i = 0; i < 6; i++) {
int taskNumber = i;
executor.submit(() -> {
System.out.println("Task " + taskNumber + " is running on " + Thread.currentThread().getName());
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
});
}
executor.shutdown();
}
}
ExecutorService 的优化与注意事项
线程池大小的选择
线程池大小的选择对性能有重要影响。如果线程池太小，可能导致任务长时间等待，无法充分利用系统资源；如果线程池太大，可能会导致过多的线程竞争资源，增加上下文切换开销，反而降低性能。
- CPU 密集型任务：对于 CPU 密集型任务，线程池大小一般设置为 CPU 核心数加 1。这样可以在一个线程因偶尔的 I/O 操作或其他原因阻塞时，仍有额外的线程可以利用 CPU 资源。
- I/O 密集型任务：对于 I/O 密集型任务，由于线程在等待 I/O 操作时会处于空闲状态，线程池大小可以设置得较大，一般为 CPU 核心数的 2 倍甚至更多，以充分利用等待 I/O 的时间来执行其他任务。
资源管理与内存泄漏
在使用 ExecutorService 时，要注意资源的正确管理。如果不及时关闭 ExecutorService，可能会导致线程无法释放，从而造成内存泄漏。另外，对于长时间运行的任务，要确保任务不会占用过多的系统资源，例如内存、文件句柄等。
异常处理
Callable 任务在执行过程中可能会抛出异常，这些异常需要正确处理。通过 Future.get() 方法获取任务结果时，如果任务执行过程中抛出异常，get() 方法会将异常包装成 ExecutionException 或 InterruptedException 抛出。在捕获这些异常时，要注意获取原始的异常信息，以便进行有效的调试和处理。
import java.util.concurrent.*;
public class ExceptionHandlingExample {
public static void main(String[] args) {
ExecutorService executorService = Executors.newSingleThreadExecutor();
Future<Integer> future = executorService.submit(() -> {
// 模拟可能抛出异常的任务
if (Math.random() < 0.5) {
throw new RuntimeException("Task failed");
}
return 42;
});
try {
System.out.println("Task result: " + future.get());
} catch (InterruptedException | ExecutionException e) {
if (e.getCause() != null) {
System.out.println("Original exception: " + e.getCause().getMessage());
} else {
e.printStackTrace();
}
}
executorService.shutdown();
}
}
总结
ExecutorService 是 Java 并发编程中一个强大的工具，它提供了丰富的特性来管理和控制异步任务的执行。通过合理使用 ExecutorService，可以提高程序的性能、响应速度，并有效地管理系统资源。在实际应用中，需要根据具体的业务场景选择合适的线程池类型、任务队列和饱和策略，同时注意线程池大小的优化、资源管理以及异常处理等方面，以确保程序的稳定性和高效性。希望通过本文的介绍，读者能够深入理解 ExecutorService 的特性和使用方法，并在实际项目中灵活运用。