Java并行流与ExecutorService：深度解析并发任务执行机制

import java.util.Set;
import java.util.concurrent.TimeUnit;

public class ParallelStreamDemo {

    // 模拟一个耗时任务
    private static Runnable heavyTask(String taskId) {
        return () -> {
            try {
                System.out.println(Thread.currentThread().getName() + " executing " + taskId);
                TimeUnit.MILLISECONDS.sleep(500); // 模拟耗时操作
                System.out.println(Thread.currentThread().getName() + " finished " + taskId);
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                System.err.println(Thread.currentThread().getName() + " interrupted during " + taskId);
            }
        };
    }

    public static void main(String[] args) {
        Set<Runnable> tasks = Set.of(
            heavyTask("Task A"), heavyTask("Task B"), 
            heavyTask("Task C"), heavyTask("Task D"),
            heavyTask("Task E"), heavyTask("Task F"),
            heavyTask("Task G"), heavyTask("Task H")
        );

        System.out.println("--- Executing with parallelStream() ---");
        tasks.parallelStream().forEach(Runnable::run);
        System.out.println("--- parallelStream() execution finished ---");
    }
}

import java.util.Set;
import java.util.concurrent.*;

public class ExecutorServiceDemo {

    // 模拟一个耗时任务
    private static Callable<Object> heavyTask(String taskId) {
        return () -> {
            try {
                System.out.println(Thread.currentThread().getName() + " executing " + taskId);
                TimeUnit.MILLISECONDS.sleep(500); // 模拟耗时操作
                System.out.println(Thread.currentThread().getName() + " finished " + taskId);
                return "Completed " + taskId;
            } catch (InterruptedException e) {
                Thread.currentThread().interrupt();
                System.err.println(Thread.currentThread().getName() + " interrupted during " + taskId);
                throw new RuntimeException("Task interrupted", e);
            }
        };
    }

    public static void main(String[] args) {
        Set<Callable<Object>> tasks = Set.of(
            heavyTask("Task A"), heavyTask("Task B"), 
            heavyTask("Task C"), heavyTask("Task D"),
            heavyTask("Task E"), heavyTask("Task F"),
            heavyTask("Task G"), heavyTask("Task H")
        );

        System.out.println("--- Executing with ExecutorService ---");
        // 创建一个固定大小的线程池，例如4个线程
        ExecutorService executor = Executors.newFixedThreadPool(4); 
        try {
            // 提交所有任务并等待它们完成
            executor.invokeAll(tasks).forEach(future -> {
                try {
                    future.get(); // 获取任务结果，等待任务完成
                } catch (InterruptedException | ExecutionException e) {
                    System.err.println("Task execution failed: " + e.getMessage());
                }
            });
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            System.err.println("Main thread interrupted while invoking tasks.");
        } finally {
            // 务必关闭ExecutorService，释放资源
            executor.shutdown(); 
            try {
                if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
                    executor.shutdownNow(); // 强制关闭
                }
            } catch (InterruptedException e) {
                executor.shutdownNow();
                Thread.currentThread().interrupt();
            }
        }
        System.out.println("--- ExecutorService execution finished ---");
    }
}