Java并发编程：使用ExecutorService限制并发线程数量

import com.google.gson.Gson;
import java.io.FileWriter;
import java.io.IOException;
import java.io.Writer;
import java.nio.file.Path; // 假设有一个路径来存储文件

// 假设 EventuelleDestination, EmployeDao, EntrepriseDao 是已定义的类
// 这里为了简化，我们只关注序列化逻辑
class EventuelleDestination {
    private String id; // 假设有一个ID
    private String name; // 假设有一个名称
    // ... 其他属性和getter/setter
    public EventuelleDestination(String id, String name) {
        this.id = id;
        this.name = name;
    }
    public String getId() { return id; }
    public String getName() { return name; }
    @Override
    public String toString() { return "EventuelleDestination{" + "id='" + id + '\'' + ", name='" + name + '\'' + '}'; }
}

class SerializationTask implements Runnable {
    private final EventuelleDestination destination;
    private final Path outputDirectory; // 输出目录

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

    @Override
    public void run() {
        // 模拟原始的序列化逻辑
        // 实际应用中，employeDao和entrepriseDao应通过依赖注入或其他方式获取
        // 这里为了演示，我们简化文件名生成
        String filename = "/" + destination.getId() + "_" + destination.getName() + ".json";

        try (Writer writer = new FileWriter(outputDirectory.resolve(filename).toFile())) {
            new Gson().toJson(destination, writer);
            System.out.println(Thread.currentThread().getName() + " - " + destination + " has been serialized...");
        } catch (IOException e) {
            System.err.println(Thread.currentThread().getName() + " - Error serializing " + destination + ": " + e.getMessage());
            e.printStackTrace();
        }
    }
}

import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class ThreadPoolDemo {

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

    private void runSerializationDemo() {
        // 1. 准备任务数据
        List<EventuelleDestination> destinations = new ArrayList<>();
        for (int i = 0; i < 20; i++) {
            destinations.add(new EventuelleDestination("ID_" + i, "Location_" + i));
        }

        // 2. 指定输出目录
        Path outputDir = Paths.get("serialized_data");
        try {
            Files.createDirectories(outputDir); // 确保目录存在
        } catch (IOException e) {
            System.err.println("Failed to create output directory: " + e.getMessage());
            return;
        }

        // 3. 创建固定大小的线程池，限制同时运行3个线程
        // 这里的3是根据原始问题中的需求设定的
        ExecutorService executorService = Executors.newFixedThreadPool(3);
        System.out.println("ExecutorService created with 3 threads.");

        // 4. 提交所有任务到线程池
        for (EventuelleDestination dest : destinations) {
            executorService.submit(new SerializationTask(dest, outputDir));
        }

        // 5. 优雅地关闭线程池
        shutdownAndAwaitTermination(executorService);
        System.out.println("All serialization tasks completed.");
    }

    /**
     * 优雅地关闭ExecutorService，等待所有任务完成。
     * 这是来自Javadoc的推荐模式。
     * @param executorService 要关闭的ExecutorService
     */
    void shutdownAndAwaitTermination(ExecutorService executorService) {
        executorService.shutdown(); // 禁用新任务提交

        try {
            // 等待现有任务在指定时间内终止
            if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
                executorService.shutdownNow(); // 如果超时，取消当前正在执行的任务
                // 再次等待，确保任务对取消做出响应
                if (!executorService.awaitTermination(60, TimeUnit.SECONDS)) {
                    System.err.println("ExecutorService did not terminate.");
                }
            }
        } catch (InterruptedException ie) {
            // (重新)取消如果当前线程也被中断
            executorService.shutdownNow();
            // 保留中断状态
            Thread.currentThread().interrupt();
        }
    }
}