Java多线程ThreadPoolExecutor详解

ThreadPoolExecutor 是 Java 中常用的线程池实现类，通过线程池可以更好地使用资源，提高程序性能。本文将详细讲解 ThreadPoolExecutor 的使用，包括线程池的创建、使用和销毁等方面。

线程池的创建

线程池是通过 ThreadPoolExecutor 类创建的，构造方法有几个重载，可以根据需求选择使用。

常用构造方法

public ThreadPoolExecutor(int corePoolSize, 
                           int maximumPoolSize, 
                           long keepAliveTime, 
                           TimeUnit unit, 
                           BlockingQueue<Runnable> workQueue)

该构造方法是最常用的，它会创建一个新的线程池对象，其中：

• corePoolSize：核心线程数，即线程池中最少会保持的活跃线程数。
• maximumPoolSize：最大线程数，即线程池中最多会同时存在的线程数。
• keepAliveTime：线程空闲时间，超过该时间的线程会被销毁。
• TimeUnit：时间单位，用于指定时间的单位，如 TimeUnit.SECONDS 为秒。
• BlockingQueue<Runnable>：任务队列，用于存储线程池中等待执行的任务。

其他构造方法

还有其他的构造方法，包括使用默认参数创建线程池的无参构造方法、可以指定线程工厂的构造方法、可以指定拒绝策略的构造方法等。这里不做详细说明，可参考官方文档。

线程池的使用

线程池创建完成之后，下一步是将任务提交到线程池中执行。有三种提交任务的方法：`

execute() 方法

execute() 方法用于异步提交一个任务，提交后立即返回，不会等待任务执行完成，示例如下：

ExecutorService executor = Executors.newFixedThreadPool(2);
executor.execute(() -> {
    System.out.println("Task 1 is running.");
});

submit() 方法

submit() 方法除了异步提交任务，还会返回一个 Future 实例，可以通过它获取任务的执行结果，示例如下：

ExecutorService executor = Executors.newFixedThreadPool(2);
Future<String> future = executor.submit(() -> {
    System.out.println("Task 1 is running.");
    return "Task 1";
});
System.out.println(future.get()); // 输出：Task 1

invokeAll() 方法

invokeAll() 方法用于批量提交任务，并等待所有任务执行完成，示例如下：

ExecutorService executor = Executors.newFixedThreadPool(2);
List<Callable<String>> tasks = new ArrayList<>();
tasks.add(() -> {
    System.out.println("Task 1 is running.");
    return "Task 1";
});
tasks.add(() -> {
    System.out.println("Task 2 is running.");
    return "Task 2";
});
List<Future<String>> futures = executor.invokeAll(tasks);
futures.forEach(future -> {
    try {
        System.out.println(future.get());
    } catch (InterruptedException | ExecutionException e) {
        e.printStackTrace();
    }
});

线程池的销毁

线程池有两种销毁方法，分别是 shutdown() 和 shutdownNow()。

shutdown() 方法

调用 shutdown() 方法后，线程池会尝试将等待执行的任务全部执行完毕，不再接受新的任务提交，示例如下：

ExecutorService executor = Executors.newFixedThreadPool(2);
executor.shutdown();

shutdownNow() 方法

调用 shutdownNow() 方法后，线程池会立即关闭，不会等待等待执行的任务全部执行完毕，示例如下：

ExecutorService executor = Executors.newFixedThreadPool(2);
executor.shutdownNow();

除此之外，还可以通过其他方法获取线程池状态、设置线程池大小、修改拒绝策略等等。

示例说明

下面是两条使用 ThreadPoolExecutor 的示例说明。

示例一

有一个消息队列，需要不断地往队列中插入消息，同时有多个工作线程从队列中读取消息并进行处理。为了避免每次都创建线程，我们可以使用线程池来管理这些工作线程。具体实现如下：

public class MessageQueue {
    private BlockingQueue<String> queue = new LinkedBlockingQueue<>();
    private ThreadPoolExecutor executor = new ThreadPoolExecutor(
            5, 10, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100));

    public void addMessage(String message) {
        queue.add(message);
    }

    public void start() {
        while (true) {
            try {
                String message = queue.take();
                executor.execute(() -> {
                    // 处理消息
                });
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }

    public void stop() {
        executor.shutdownNow();
    }
}

该示例使用 BlockingQueue 存储消息，使用 ThreadPoolExecutor 管理工作线程。在 start() 方法中，通过 BlockingQueue 不断地从队列中提取消息并将其作为任务提交到线程池中执行。在 stop() 方法中，调用 shutdownNow() 方法关闭线程池。

示例二

有一个需要处理的任务列表，需要多个工作线程并发处理，每一个任务的执行时间都不一样。为了使用多个线程并发执行任务，我们可以使用线程池来管理这些工作线程。具体实现如下：

public class TaskList {
    private List<Task> taskList = new ArrayList<>();
    private ThreadPoolExecutor executor = new ThreadPoolExecutor(
            3, 6, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(100));

    public void addTask(Task task) {
        taskList.add(task);
    }

    public void start() {
        CountDownLatch latch = new CountDownLatch(taskList.size());
        List<Future<?>> futures = new ArrayList<>();
        for (Task task : taskList) {
            futures.add(executor.submit(() -> {
                try {
                    task.execute();
                } finally {
                    latch.countDown();
                }
            }));
        }
        try {
            latch.await();
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        futures.forEach(future -> {
            try {
                future.get();
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        });
    }

    public void stop() {
        executor.shutdownNow();
    }
}

该示例使用 List 存储任务列表，使用 ThreadPoolExecutor 管理工作线程。在 start() 方法中，通过 ExecutorService 提交每一个任务到线程池中执行，并使用 CountDownLatch 检测所有任务是否执行完毕。在 stop() 方法中，调用 shutdownNow() 方法关闭线程池。

以上就是 ThreadPoolExecutor 的详细使用攻略，包括线程池的创建、使用和销毁等方面，并提供了两个示例说明。希望对您有所帮助。