在ExecutorService中实现可控的任务中断与取消

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;

public class ExecutorServiceTaskCancellation {

    // 模拟一个执行耗时任务的方法
    private static void doTask(Object obj) {
        try {
            System.out.println(Thread.currentThread().getName() + " - 开始处理: " + obj);
            // 模拟任务执行，并定期检查中断状态
            for (int i = 0; i < 10; i++) {
                if (Thread.interrupted()) { // 检查中断标志
                    System.out.println(Thread.currentThread().getName() + " - 任务 " + obj + " 被中断，提前退出。");
                    return; // 响应中断，退出任务
                }
                Thread.sleep(500); // 模拟耗时操作
            }
            System.out.println(Thread.currentThread().getName() + " - 完成处理: " + obj);
        } catch (InterruptedException e) {
            // 捕获InterruptedException，同样表示中断
            System.out.println(Thread.currentThread().getName() + " - 任务 " + obj + " 捕获到InterruptedException，提前退出。");
            // 重新设置中断标志，因为捕获InterruptedException会清除中断标志
            Thread.currentThread().interrupt();
        }
    }

    public static void main(String[] args) {
        // 创建一个固定大小的线程池，例如5个线程
        ExecutorService executor = Executors.newFixedThreadPool(5);

        // 模拟一个大的对象列表
        List<String> objectList = new ArrayList<>();
        for (int i = 0; i < 20; i++) {
            objectList.add("Task-" + (i + 1));
        }

        // 将对象列表分成每组5个
        // 假设这里使用Guava的Lists.partition，实际项目中请引入Guava库
        // 或者自己实现分区逻辑
        List<List<String>> objectGroups = partitionList(objectList, 5);

        int groupCount = 0;
        for (List<String> eachGroup : objectGroups) {
            groupCount++;
            System.out.println("\n--- 开始处理第 " + groupCount + " 组任务 ---");

            CountDownLatch latch = new CountDownLatch(eachGroup.size());
            List<Future<?>> futures = new ArrayList<>(); // 存储当前组的所有Future对象

            for (String obj : eachGroup) {
                Future<?> future = executor.submit(() -> {
                    try {
                        doTask(obj);
                    } finally {
                        latch.countDown(); // 无论任务成功、失败或中断，都减少计数
                    }
                });
                futures.add(future); // 将Future添加到列表中
            }

            try {
                // 等待当前组任务完成，最长等待15分钟
                if (!latch.await(15, TimeUnit.SECONDS)) { // 将15分钟改为15秒方便测试
                    System.out.println("警告：第 " + groupCount + " 组任务在15秒内未能全部完成，尝试取消未完成任务。");
                    // 超时发生，尝试取消所有尚未完成的任务
                    for (Future<?> future : futures) {
                        if (!future.isDone()) { // 检查任务是否已经完成
                            boolean cancelled = future.cancel(true); // 尝试中断任务
                            System.out.println("  - 尝试取消任务: " + (cancelled ? "成功" : "失败或已完成") + " for " + future);
                        }
                    }
                } else {
                    System.out.println("第 " + groupCount + " 组任务全部完成。");
                }
            } catch (InterruptedException e) {
                System.out.println("等待第 " + groupCount + " 组任务时被中断: " + e.getMessage());
                Thread.currentThread().interrupt(); // 重新设置中断标志
            }
        }

        // 所有组处理完毕后，优雅地关闭ExecutorService
        System.out.println("\n所有任务组处理完毕，准备关闭ExecutorService。");
        executor.shutdown(); // 阻止新任务提交，等待已提交任务完成
        try {
            if (!executor.awaitTermination(60, TimeUnit.SECONDS)) {
                System.err.println("ExecutorService未能及时关闭，尝试强制关闭。");
                executor.shutdownNow(); // 强制关闭
            }
        } catch (InterruptedException e) {
            System.err.println("关闭ExecutorService时被中断: " + e.getMessage());
            executor.shutdownNow(); // 捕获中断异常时，也强制关闭
            Thread.currentThread().interrupt();
        }
        System.out.println("ExecutorService已关闭。");
    }

    // 模拟Lists.partition方法，实际项目中可使用Guava库
    private static <T> List<List<T>> partitionList(List<T> list, int size) {
        List<List<T>> partitions = new ArrayList<>();
        if (list == null || list.isEmpty() || size <= 0) {
            return partitions;
        }
        for (int i = 0; i < list.size(); i += size) {
            partitions.add(new ArrayList<>(list.subList(i, Math.min(i + size, list.size())));
        }
        return partitions;
    }
}