Java ExecutorService：高效管理并发任务与线程限制

本教程详细介绍了如何使用java的executors框架，特别是`executorservice`和`executors.newfixedthreadpool()`方法，来有效管理并发任务并限制同时运行的线程数量。通过将每个任务封装为`runnable`或`callable`，并提交给固定大小的线程池，开发者可以确保系统资源被合理利用，避免因过多线程导致的性能问题，并学习如何优雅地关闭线程池。

引言：并发任务管理挑战

在现代应用程序开发中，处理大量独立但耗时的任务是一个常见需求，例如批量数据处理、文件序列化、网络请求等。直接为每个任务创建一个新线程虽然简单，但会导致严重的性能问题和资源耗尽。操作系统和JVM管理大量线程的开销巨大，过多的线程切换（上下文切换）会降低整体吞吐量。因此，限制同时运行的线程数量，实现任务的并发但不并行，是优化资源利用率和系统稳定性的关键。Java的java.util.concurrent包，特别是Executors框架，为解决这一挑战提供了强大而优雅的解决方案。

Java Executors 框架简介

Java 5引入的Executors框架极大地简化了多线程编程。它提供了一套高级API，用于管理线程池和任务提交，将任务的提交与任务的执行解耦。开发者无需手动创建和管理线程，只需定义好任务并将其提交给ExecutorService，后者会负责调度和执行这些任务。

定义并发任务：Runnable 与 Callable

在Executors框架中，并发任务通常通过实现Runnable或Callable接口来定义。

• Runnable接口：适用于不需要返回结果，也不抛出受检查异常的任务。它只包含一个run()方法。
• Callable接口：适用于需要返回结果，并且可能抛出受检查异常的任务。它包含一个call()方法，返回一个泛型结果，并可以抛出Exception。

对于需要对对象列表进行序列化这类任务，如果不需要立即获取每个序列化操作的返回结果，Runnable是一个合适的选择。

立即学习“Java免费学习笔记（深入）”；

以下是根据原始问题中的序列化逻辑，将其适配为Runnable任务的示例：

Magician

Figma插件，AI生成图标、图片和UX文案

下载

import com.google.gson.Gson; // 需要添加Gson库依赖，例如：com.google.code.gsongson2.10.1
import java.io.FileWriter;
import java.io.IOException;
import java.io.Writer;
import java.nio.file.Path;
import java.time.Instant;
import java.util.concurrent.ThreadLocalRandom;

// 模拟待序列化的数据对象
class EventuelleDestination {
    private int id;
    private String name;

    public EventuelleDestination(int id, String name) {
        this.id = id;
        this.name = name;
    }

    public int getId() { return id; }
    public String getName() { return name; }

    @Override
    public String toString() { return "Destination[id=" + id + ", name=" + name + "]"; }
}

// 模拟数据访问对象（DAO）或其他依赖
class MockDao {
    public int getEmployeId() { return 1001; }
    public int getEmplacementIdByDepartementId(int deptId) { return deptId + 2000; }
}

// 定义序列化任务，实现Runnable接口
class SerializationTask implements Runnable {
    private final EventuelleDestination destination;
    private final Path outputDirectory;
    private final MockDao mockDao; // 传入必要的依赖

    public SerializationTask(EventuelleDestination destination, Path outputDirectory, MockDao mockDao) {
        this.destination = destination;
        this.outputDirectory = outputDirectory;
        this.mockDao = mockDao;
    }

    @Override
    public void run() {
        Gson gson = new Gson();
        // 简化文件名生成逻辑，实际应用中根据业务需求构建
        String filename = String.format("/%d_%d_%d.json",
                mockDao.getEmployeId(),
                mockDao.getEmplacementIdByDepartementId(destination.getId()),
                destination.getId());

        try (Writer writer = new FileWriter(outputDirectory.toString() + filename)) {
            gson.toJson(destination, writer);
            System.out.println(Thread.currentThread().getName() + " 序列化完成: " + destination.getName() + " (ID: " + destination.getId() + ") 于 " + Instant.now());
            // 模拟耗时操作
            Thread.sleep(ThreadLocalRandom.current().nextInt(50, 200));
        } catch (IOException e) {
            System.err.println("序列化 " + destination.getName() + " 失败: " + e.getMessage());
            e.printStackTrace();
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt(); // 恢复中断状态
            System.err.println(Thread.currentThread().getName() + " 任务被中断: " + destination.getName());
            throw new RuntimeException("序列化任务被中断", e);
        }
    }
}

在上述SerializationTask中，我们将原始的序列化逻辑封装在run()方法内。为了使任务独立运行，所有必要的上下文信息（如EventuelleDestination对象、输出目录、DAO依赖）都通过构造函数传入。

使用 ExecutorService 创建固定大小线程池

ExecutorService是Executors框架的核心接口，它提供了管理线程池生命周期和提交任务的方法。Executors是一个工厂类，提供了多种静态方法来创建不同类型的ExecutorService实例。

要限制同时运行的线程数量，最常用的方法是使用Executors.newFixedThreadPool(int nThreads)。这个方法会创建一个固定大小的线程池，池中线程的数量始终保持为nThreads。当有新任务提交时，如果所有线程都在忙碌，任务将被放入一个等待队列中，直到有空闲线程可用。

示例：限制并发序列化任务

下面是一个完整的示例，演示如何使用Executors.newFixedThreadPool()来限制并发序列化任务的数量为3个。

import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.List;
import java.util.ArrayList;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.stream.IntStream;
import java.io.IOException; // 导入IOException

public class ThreadPoolSerializationDemo {

    private final MockDao mockDao = new MockDao();
    private final Path outputDir = Paths.get("serialized_data"); // 定义输出目录

    public static void main(String[] args) {
        ThreadPoolSerializationDemo demo = new ThreadPoolSerializationDemo();
        demo.runDemo();
    }

    public