Java线程控制：ExecutorService管理方法

IT行业相对于一般传统行业，发展更新速度更快，一旦停止了学习，很快就会被行业所淘汰。所以我们需要踏踏实实的不断学习，精进自己的技术，尤其是初学者。今天golang学习网给大家整理了《Java并发编程：ExecutorService控制线程数方法》，聊聊，我们一起来看看吧！

本文详细介绍了在Java中如何利用`Executors`框架，特别是`ExecutorService`和`Executors.newFixedThreadPool()`方法，来有效地限制同时运行的线程数量。通过将任务封装为`Runnable`或`Callable`，并提交给固定大小的线程池，开发者可以精确控制并发度，从而优化资源使用和系统性能。文章提供了完整的代码示例，并强调了线程池的正确关闭机制。

在多线程编程中，我们经常需要处理一系列独立的任务，但又希望限制同时执行的任务数量，以避免过度消耗系统资源或造成性能瓶颈。例如，当需要对一个包含大量对象的列表进行并发序列化操作时，如果为每个对象都创建一个新线程，可能会导致系统因线程过多而崩溃。Java 5引入的Executors框架为解决此类并发问题提供了强大而简洁的工具。

任务定义：Runnable与Callable

在将任务提交给线程池执行之前，首先需要将任务逻辑封装起来。Java提供了两个核心接口用于定义并发任务：

1. Runnable:

   • 定义了一个不返回任何结果，也不抛出受检查异常的任务。
   • 其核心方法是 public void run()。
   • 适用于执行不需要返回结果的异步操作。

2. Callable:

   • 定义了一个可以返回结果，并可能抛出受检查异常的任务。
   • 其核心方法是 public T call() throws Exception。
   • 适用于需要获取任务执行结果或处理特定异常的场景，通常与 Future 结合使用。

根据原始问题中对EventuelleDestination对象进行序列化的需求，我们可以将其封装为一个Runnable任务。为了使示例完整和可运行，我们创建了一些模拟类。

import com.google.gson.Gson;
import java.io.FileWriter;
import java.io.IOException;
import java.io.Writer;
import java.nio.file.Path;
import java.util.Objects;

// 模拟的业务实体和DAO层，用于使SerializationTask独立可运行
class EventuelleDestination {
    private String name;
    private EventuelAcceuillant eventuelAcceuillant;

    public EventuelleDestination(String name, EventuelAcceuillant acceuillant) {
        this.name = name;
        this.eventuelAcceuillant = acceuillant;
    }
    public EventuelAcceuillant getEventuelAcceuillant() { return eventuelAcceuillant; }
    @Override
    public String toString() { return "EventuelleDestination{" + "name='" + name + '\'' + '}'; }
}

class EventuelAcceuillant {
    private int id;
    public EventuelAcceuillant(int id) { this.id = id; }
    public int getId() { return id; }
}

class EmployeDao {
    public Employe getEmploye() { return new Employe(1001); } // 模拟获取员工
}

class Employe {
    private int id;
    public Employe(int id) { this.id = id; }
    public int getId() { return id; }
}

class EntrepriseDao {
    public int retrouveEmplacementIdParDepartementId(int deptId) { return deptId * 10; } // 模拟获取位置ID
}

/**
 * 负责将EventuelleDestination对象序列化到文件的Runnable任务。
 */
public class SerializationTask implements Runnable {
    private final EventuelleDestination eventuelleDestination;
    private final Path dossierSoumissions; // 序列化输出的基础目录
    private final EmployeDao employeDao;
    private final EntrepriseDao entrepriseDao;

    public SerializationTask(EventuelleDestination e, Path dossierSoumissions, EmployeDao employeDao, EntrepriseDao entrepriseDao) {
        this.eventuelleDestination = Objects.requireNonNull(e, "EventuelleDestination cannot be null");
        this.dossierSoumissions = Objects.requireNonNull(dossierSoumissions, "DossierSoumissions path cannot be null");
        this.employeDao = Objects.requireNonNull(employeDao, "EmployeDao cannot be null");
        this.entrepriseDao = Objects.requireNonNull(entrepriseDao, "EntrepriseDao cannot be null");
    }

    @Override
    public void run() {
        Gson gson = new Gson();
        // 根据业务逻辑构建文件名
        String filename = "/" + employeDao.getEmploye().getId() + "_" +
                          entrepriseDao.retrouveEmplacementIdParDepartementId(eventuelleDestination.getEventuelAcceuillant().getId()) + "_" +
                          eventuelleDestination.getEventuelAcceuillant().getId() + ".json";

        try (Writer writer = new FileWriter(dossierSoumissions.resolve(filename).toString())) {
            gson.toJson(eventuelleDestination, writer);
            System.out.println(Thread.currentThread().getName() + ": " + eventuelleDestination + " 已序列化到 " + filename + "...");
        } catch (IOException e) {
            System.err.println(Thread.currentThread().getName() + ": 序列化 " + eventuelleDestination + " 时发生错误: " + e.getMessage());
            e.printStackTrace();
        }
    }
}

使用ExecutorService管理线程池

ExecutorService是Executors框架的核心接口，它提供了一套用于管理和执行提交任务的机制。Executors工具类则提供了多种静态工厂方法来创建不同类型的ExecutorService实例。

为了实现固定数量的并发线程，我们使用Executors.newFixedThreadPool(int nThreads)方法。这个方法会创建一个拥有固定线程数量的线程池。当有新任务提交时，如果池中的线程数少于nThreads，则会创建一个新线程来执行任务；如果线程数已达到nThreads，则新任务会被放入等待队列，直到池中有空闲线程可用。

下面是使用newFixedThreadPool来限制并发序列化任务的示例：

import java.nio.file.Files;
import java.nio.file.Path;
import java.time.Instant;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.io.IOException;

public class FixedThreadPoolDemo {

    private final Path tempDir; // 用于存储序列化文件的临时目录
    private final EmployeDao employeDao = new EmployeDao();
    private final EntrepriseDao entrepriseDao = new EntrepriseDao();

    public FixedThreadPoolDemo() throws IOException {
        // 创建一个临时目录用于序列化输出，确保示例的整洁性
        this.tempDir = Files.createTempDirectory("serialization_output");
        System.out.println("序列化输出目录: " + tempDir.toAbsolutePath());
    }

    public void runDemo() {
        List<EventuelleDestination> destinations = new ArrayList<>();
        // 填充一些模拟数据，共10个任务
        for (int i = 1; i <= 10; i++) {
            destinations.add(new EventuelleDestination("Destination_" + i, new EventuelAcceuillant(i)));
        }

        // 定义固定线程池的大小，这里设置为3，与问题要求一致
        final int THREAD_POOL_SIZE = 3;
        ExecutorService executorService = Executors.newFixedThreadPool(THREAD_POOL_SIZE);
        System.out.println("开始使用固定大小为 " + THREAD_POOL_SIZE + " 的线程池进行序列化。");

        // 遍历列表，将每个序列化任务提交给线程池
        for (EventuelleDestination dest : destinations) {
            executorService.submit(new SerializationTask(dest, tempDir, employeDao, entrepriseDao));
        }

        // 优雅地关闭线程池
        shutdownAndAwaitTermination(executorService);
        System.out.println("所有序列化任务已完成或终止。输出文件位于: " + tempDir.toAbsolutePath());

        // 清理临时目录（可选）
        try {
            Files.walk(tempDir)
                 .sorted(java.util.Comparator.reverseOrder()) // 先删除文件，再删除空目录
                 .map(Path::toFile)
                 .forEach(java.io.File::delete);
            Files.delete(tempDir);
            System.out.println("已清理临时目录: " + tempDir.toAbsolutePath());
        } catch (IOException e) {
            System.err.println("清理临时目录时发生错误: " + e.getMessage());
        }
    }

    /**
     * 优雅地关闭ExecutorService的工具方法。
     * 参照JavaDoc中的最佳实践。
     */
    void shutdownAndAwaitTermination(ExecutorService pool) {
        pool.shutdown(); // 停止接收新任务
        try {
            // 等待已提交任务完成，最多等待60秒
            if (!pool.awaitTermination(60, TimeUnit.SECONDS)) {
                pool.shutdownNow(); // 强制取消当前正在执行的任务
                // 再次等待，确保任务响应中断
                if (!pool.awaitTermination(60, TimeUnit.SECONDS)) {
                    System.err.println("执行器服务未能终止。 " + Instant.now());
                }
            }
        } catch (InterruptedException ex) {
            // 如果当前线程在等待期间被中断，则重新取消所有任务
            pool.shutdownNow();
            // 保留中断状态
            Thread.currentThread().interrupt

好了，本文到此结束，带大家了解了《Java线程控制：ExecutorService管理方法》，希望本文对你有所帮助！关注golang学习网公众号，给大家分享更多文章知识！