跳转至

Java中的Executor框架：深入理解与实践

简介

在Java多线程编程中，Executor框架是一个强大的工具，它简化了线程的管理和任务的执行。通过使用Executor框架，我们可以更高效地控制线程池的创建、任务的提交与执行，从而提升应用程序的性能和可维护性。本文将详细介绍Executor框架的基础概念、使用方法、常见实践以及最佳实践，并通过丰富的代码示例帮助读者深入理解。

目录

  1. 基础概念
  2. 使用方法
    • 2.1 创建线程池
    • 2.2 提交任务
    • 2.3 关闭线程池
  3. 常见实践
    • 3.1 固定大小线程池
    • 3.2 缓存线程池
    • 3.3 单线程线程池
    • 3.4 定时任务线程池
  4. 最佳实践
    • 4.1 合理设置线程池大小
    • 4.2 处理任务异常
    • 4.3 监控线程池状态
  5. 小结
  6. 参考资料

基础概念

Executor是Java并发包中的一个接口，它定义了一个用于执行任务的简单方法：

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

这个接口的目的是将任务的提交和执行分离。通过实现Executor接口，我们可以定义不同的任务执行策略。

ExecutorService是Executor的子接口，它提供了更丰富的方法来管理任务的执行，例如关闭线程池、提交返回结果的任务等。

ThreadPoolExecutor是ExecutorService的一个实现类，它是线程池的核心实现，负责管理线程的创建、销毁和任务的调度。

使用方法

创建线程池

创建线程池可以使用Executors类提供的工厂方法，也可以直接使用ThreadPoolExecutor的构造函数。

使用Executors工厂方法

  • 固定大小线程池：
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class FixedThreadPoolExample {
    public static void main(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(3);
        for (int i = 0; i < 5; i++) {
            final int taskNumber = i;
            executorService.submit(() -> {
                System.out.println("Task " + taskNumber + " is running on thread " + Thread.currentThread().getName());
            });
        }
        executorService.shutdown();
    }
}

在这个示例中，我们创建了一个固定大小为3的线程池。这意味着线程池最多同时执行3个任务，其余任务将在队列中等待。

  • 缓存线程池：
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class CachedThreadPoolExample {
    public static void main(String[] args) {
        ExecutorService executorService = Executors.newCachedThreadPool();
        for (int i = 0; i < 5; i++) {
            final int taskNumber = i;
            executorService.submit(() -> {
                System.out.println("Task " + taskNumber + " is running on thread " + Thread.currentThread().getName());
            });
        }
        executorService.shutdown();
    }
}

缓存线程池会根据需要创建新线程，如果有空闲线程则会复用。

  • 单线程线程池：
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

public class SingleThreadExecutorExample {
    public static void main(String[] args) {
        ExecutorService executorService = Executors.newSingleThreadExecutor();
        for (int i = 0; i < 5; i++) {
            final int taskNumber = i;
            executorService.submit(() -> {
                System.out.println("Task " + taskNumber + " is running on thread " + Thread.currentThread().getName());
            });
        }
        executorService.shutdown();
    }
}

单线程线程池只有一个线程，所有任务将按顺序执行。

  • 定时任务线程池：
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

public class ScheduledThreadPoolExample {
    public static void main(String[] args) {
        ScheduledExecutorService scheduledExecutorService = Executors.newScheduledThreadPool(2);
        scheduledExecutorService.scheduleAtFixedRate(() -> {
            System.out.println("Scheduled task is running on thread " + Thread.currentThread().getName());
        }, 0, 2, TimeUnit.SECONDS);
    }
}

定时任务线程池可以按照固定的时间间隔或延迟执行任务。

直接使用ThreadPoolExecutor构造函数

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class CustomThreadPoolExample {
    public static void main(String[] args) {
        BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>(10);
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                2,
                4,
                1,
                TimeUnit.MINUTES,
                queue);
        for (int i = 0; i < 15; i++) {
            final int taskNumber = i;
            executor.submit(() -> {
                System.out.println("Task " + taskNumber + " is running on thread " + Thread.currentThread().getName());
            });
        }
        executor.shutdown();
    }
}

在这个示例中，我们创建了一个自定义的线程池，初始线程数为2，最大线程数为4，线程空闲时存活时间为1分钟，任务队列容量为10。

提交任务

ExecutorService提供了几种提交任务的方法： - submit(Runnable task)：提交一个Runnable任务，返回一个Future对象，可以通过Future对象判断任务是否完成。

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

public class SubmitTaskExample {
    public static void main(String[] args) throws Exception {
        ExecutorService executorService = Executors.newFixedThreadPool(1);
        Future<?> future = executorService.submit(() -> {
            System.out.println("Task is running on thread " + Thread.currentThread().getName());
        });
        while (!future.isDone()) {
            System.out.println("Task is still running...");
        }
        System.out.println("Task has completed.");
        executorService.shutdown();
    }
}
  • submit(Callable<T> task)：提交一个Callable任务，返回一个Future对象，可以通过Future对象获取任务的执行结果。
import java.util.concurrent.*;

public class CallableTaskExample {
    public static void main(String[] args) throws Exception {
        ExecutorService executorService = Executors.newFixedThreadPool(1);
        Future<Integer> future = executorService.submit(() -> {
            System.out.println("Callable task is running on thread " + Thread.currentThread().getName());
            return 42;
        });
        Integer result = future.get();
        System.out.println("Task result: " + result);
        executorService.shutdown();
    }
}
  • execute(Runnable task)：这是Executor接口定义的方法，直接提交一个Runnable任务，没有返回值。

关闭线程池

可以通过shutdown()和shutdownNow()方法关闭线程池。 - shutdown()：启动一个有序关闭，不再接受新任务，但会继续执行已提交的任务。

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

public class ShutdownExample {
    public static void main(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(3);
        for (int i = 0; i < 5; i++) {
            final int taskNumber = i;
            executorService.submit(() -> {
                System.out.println("Task " + taskNumber + " is running on thread " + Thread.currentThread().getName());
            });
        }
        executorService.shutdown();
        while (!executorService.isTerminated()) {
            System.out.println("Waiting for tasks to complete...");
        }
        System.out.println("All tasks have completed.");
    }
}
  • shutdownNow()：尝试停止所有正在执行的任务，停止等待任务的处理，并返回等待执行的任务列表。

常见实践

固定大小线程池

适用于已知并发量且任务执行时间相对稳定的场景，例如数据库批量操作、文件处理等。可以有效控制并发度，避免资源过度消耗。

缓存线程池

适合任务执行时间短、并发量不确定的场景，例如处理大量的HTTP请求。线程池会根据需要动态创建和回收线程，提高资源利用率。

单线程线程池

适用于需要按顺序执行任务的场景，例如对共享资源的串行访问，确保数据的一致性。

定时任务线程池

常用于需要定时执行任务的场景，如定时数据备份、定时发送邮件等。

最佳实践

合理设置线程池大小

线程池大小的设置需要根据任务的类型（CPU密集型、I/O密集型）和系统资源来确定。 - CPU密集型任务：线程池大小一般设置为CPU核心数 + 1，避免线程过多导致上下文切换开销增大。 - I/O密集型任务：可以适当增大线程池大小，一般设置为CPU核心数 * 2，充分利用CPU资源。

处理任务异常

在使用ExecutorService提交任务时，需要处理任务执行过程中抛出的异常。可以通过try-catch块包裹任务代码，或者使用Future对象的get()方法捕获异常。

import java.util.concurrent.*;

public class ExceptionHandlingExample {
    public static void main(String[] args) {
        ExecutorService executorService = Executors.newFixedThreadPool(1);
        Future<?> future = executorService.submit(() -> {
            throw new RuntimeException("Task failed!");
        });
        try {
            future.get();
        } catch (InterruptedException | ExecutionException e) {
            System.out.println("Task exception: " + e.getMessage());
        }
        executorService.shutdown();
    }
}

监控线程池状态

可以通过ThreadPoolExecutor的方法监控线程池的状态，如活动线程数、已完成任务数等。

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

public class ThreadPoolMonitoringExample {
    public static void main(String[] args) {
        BlockingQueue<Runnable> queue = new LinkedBlockingQueue<>(10);
        ThreadPoolExecutor executor = new ThreadPoolExecutor(
                2,
                4,
                1,
                TimeUnit.MINUTES,
                queue);
        for (int i = 0; i < 15; i++) {
            final int taskNumber = i;
            executor.submit(() -> {
                System.out.println("Task " + taskNumber + " is running on thread " + Thread.currentThread().getName());
            });
        }
        System.out.println("Active threads: " + executor.getActiveCount());
        System.out.println("Completed tasks: " + executor.getCompletedTaskCount());
        executor.shutdown();
    }
}

小结

Executor框架为Java多线程编程提供了强大而灵活的任务执行和线程管理能力。通过合理使用线程池、正确提交任务和关闭线程池，以及遵循最佳实践，我们可以编写高效、稳定的多线程应用程序。希望本文的介绍和示例能帮助读者更好地理解和应用Executor框架。

参考资料