本文详细介绍了在java中如何利用`executors`框架，特别是`executorservice`和`executors.newfixedthreadpool()`方法，来有效地限制同时运行的线程数量。通过将任务封装为`runnable`或`callable`，并提交给固定大小的线程池，开发者可以精确控制并发度，从而优化资源使用和系统性能。文章提供了完整的代码示例，并强调了线程池的正确关闭机制。
在多线程编程中，我们经常需要处理一系列独立的任务，但又希望限制同时执行的任务数量，以避免过度消耗系统资源或造成性能瓶颈。例如，当需要对一个包含大量对象的列表进行并发序列化操作时，如果为每个对象都创建一个新线程，可能会导致系统因线程过多而崩溃。Java 5引入的Executors框架为解决此类并发问题提供了强大而简洁的工具。
任务定义：Runnable与Callable
在将任务提交给线程池执行之前，首先需要将任务逻辑封装起来。Java提供了两个核心接口用于定义并发任务：
-
Runnable:
- 定义了一个不返回任何结果，也不抛出受检查异常的任务。
- 其核心方法是 public void run()。
- 适用于执行不需要返回结果的异步操作。
-
Callable<T>:
立即学习“Java免费学习笔记（深入）”；
- 定义了一个可以返回结果，并可能抛出受检查异常的任务。
- 其核心方法是 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 
