JAVA 并发任务超时取消不生效？ExecutorService + Future 超时控制技巧

大家好，今天我们来聊聊 Java 并发编程中一个常见但又容易踩坑的问题：ExecutorService + Future 实现的任务超时取消，有时候会失效。

我们经常会使用 ExecutorService 来提交并发任务，并通过 Future 来获取任务的结果或者控制任务的生命周期，例如设置超时时间并取消任务。然而，在实际应用中，我们可能会发现，即使设置了超时时间并调用了 Future.cancel(true)，任务仍然在后台默默执行，资源没有被释放，这会导致严重的性能问题甚至程序崩溃。

为什么会出现这种情况？如何才能正确地实现任务超时取消？今天我们就来深入探讨这些问题，并提供一些实用的技巧。

任务超时取消的基本原理

首先，让我们回顾一下 ExecutorService 和 Future 在任务超时取消中的作用。

ExecutorService 负责管理线程池和提交任务。通过 ExecutorService.submit(Callable<T> task) 或者 ExecutorService.submit(Runnable task) 可以将任务提交给线程池执行。

Future 代表异步计算的结果。它提供了方法来检查计算是否完成、等待计算完成并获取计算结果。重要的是，Future 还提供了 cancel(boolean mayInterruptIfRunning) 方法来尝试取消任务。

cancel(boolean mayInterruptIfRunning) 方法的参数 mayInterruptIfRunning 决定了取消操作的行为：

true: 尝试中断正在执行的任务。
false: 仅阻止尚未开始执行的任务执行。

理想情况下，当我们调用 future.cancel(true) 时，线程池中的线程会收到中断信号，任务会停止执行，资源会被释放。但现实往往并非如此简单。

超时取消失效的常见原因

任务超时取消失效的原因有很多，主要可以归结为以下几点：

任务本身忽略或无法响应中断信号。
- 阻塞 I/O 操作： 如果任务中包含阻塞 I/O 操作，例如网络请求或者文件读写，而这些 I/O 操作没有设置超时时间或者没有正确处理 InterruptedException，那么即使线程被中断，任务仍然会阻塞在那里，无法响应取消请求。
- 长时间的 CPU 密集型计算： 如果任务是一个 CPU 密集型计算，并且没有周期性地检查中断状态，那么即使线程被中断，任务仍然会继续执行，直到计算完成。
- 使用了锁而没有正确释放： 如果任务获取了锁，但是在线程被中断时没有正确释放锁，那么其他线程可能会一直阻塞等待该锁，导致系统资源被占用。
cancel() 方法调用时机不正确。
- 在任务已经完成或取消后调用 cancel()： Future.cancel() 方法在任务已经完成、取消或者发生异常时调用，不会产生任何效果。确保在任务还在运行状态时调用 cancel() 方法。
线程池配置不当。
- 核心线程数过高： 如果线程池的核心线程数设置得过高，那么任务可能会一直被提交到核心线程中执行，而无法被取消。
- 使用了无界队列： 如果线程池使用了无界队列，那么任务可能会一直被添加到队列中，即使线程池已经满了，也不会拒绝新的任务。这会导致任务无法及时被执行和取消。
代码逻辑错误。
- 异常处理不当： 如果在任务中捕获了 InterruptedException，但是没有正确处理，例如没有重新抛出中断异常或者没有退出任务，那么任务仍然会继续执行。
- 资源没有释放： 如果在任务中使用了资源，例如文件句柄、数据库连接等，但是在线程被中断时没有正确释放这些资源，那么这些资源可能会一直被占用，导致系统资源泄露。

如何正确实现任务超时取消

为了解决上述问题，我们需要采取一系列的措施来确保任务能够正确地被取消。

响应中断信号：

阻塞 I/O 操作： 使用带有超时时间的 I/O 操作，并捕获 InterruptedException，在捕获到中断异常时，立即退出任务。例如，在使用 Socket 进行网络请求时，可以设置 Socket.setSoTimeout() 方法来设置超时时间。
CPU 密集型计算： 在 CPU 密集型计算中，周期性地检查当前线程的中断状态，如果发现线程被中断，则立即退出计算。可以使用 Thread.currentThread().isInterrupted() 方法来检查中断状态。

public class InterruptibleTask implements Callable<String> {
    @Override
    public String call() throws Exception {
        try {
            // 模拟一个长时间的计算
            for (int i = 0; i < 1000000; i++) {
                // 检查中断状态
                if (Thread.currentThread().isInterrupted()) {
                    System.out.println("任务被中断");
                    return "任务被中断";
                }
                // 执行计算
                //...
            }
            return "任务完成";
        } catch (Exception e) {
            // 处理异常
            System.out.println("任务执行出错：" + e.getMessage());
            throw e;
        }
    }
}

正确处理 InterruptedException：

重新抛出中断异常： 如果捕获到 InterruptedException，并且无法完全处理该异常，那么应该重新抛出该异常，以便让上层调用者知道任务被中断。可以使用 Thread.currentThread().interrupt() 方法来重新设置中断状态，然后再抛出 InterruptedException。
退出任务： 在捕获到 InterruptedException 时，应该立即退出任务，释放资源，并返回一个表示任务被取消的结果。

public class InterruptibleIOTask implements Callable<String> {
    @Override
    public String call() throws Exception {
        try {
            // 模拟一个阻塞 I/O 操作
            Socket socket = new Socket("example.com", 80);
            socket.setSoTimeout(5000); // 设置超时时间
            BufferedReader reader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
            String line = reader.readLine();
            return line;
        } catch (SocketTimeoutException e) {
            System.out.println("连接超时");
            return "连接超时";
        } catch (InterruptedException e) {
            System.out.println("任务被中断");
            Thread.currentThread().interrupt(); // 重新设置中断状态
            throw e; // 重新抛出中断异常
        } catch (IOException e) {
            System.out.println("I/O 异常：" + e.getMessage());
            throw e;
        }
    }
}

使用 try-finally 块释放资源：

无论任务是正常完成、发生异常还是被中断，都应该确保资源被正确释放。可以使用 try-finally 块来确保资源在任何情况下都会被释放。

public class ResourceReleasingTask implements Callable<String> {
    private FileOutputStream outputStream;

    @Override
    public String call() throws Exception {
        try {
            outputStream = new FileOutputStream("output.txt");
            // 执行写入操作
            // ...
            return "任务完成";
        } catch (IOException e) {
            System.out.println("I/O 异常：" + e.getMessage());
            throw e;
        } finally {
            // 确保资源被释放
            if (outputStream != null) {
                try {
                    outputStream.close();
                } catch (IOException e) {
                    System.out.println("关闭文件输出流失败：" + e.getMessage());
                }
            }
        }
    }
}

选择合适的线程池配置：
- 合理设置核心线程数和最大线程数： 根据任务的类型和数量，合理设置线程池的核心线程数和最大线程数。如果任务是 I/O 密集型，可以适当增加线程数；如果任务是 CPU 密集型，则应该减少线程数，避免过多的线程竞争 CPU 资源。
- 使用有界队列： 使用有界队列可以防止任务过多地堆积在队列中，导致系统资源被耗尽。可以使用 ArrayBlockingQueue 或者 LinkedBlockingQueue 来创建有界队列。
- 使用 ThreadPoolExecutor 的 rejectedExecutionHandler： 当线程池中的线程和队列都满了时，ThreadPoolExecutor 会调用 rejectedExecutionHandler 来处理被拒绝的任务。可以自定义 rejectedExecutionHandler 来记录日志、抛出异常或者执行其他操作。
```
ExecutorService executor = new ThreadPoolExecutor(
        5, // 核心线程数
        10, // 最大线程数
        60L, TimeUnit.SECONDS, // 线程空闲时间
        new ArrayBlockingQueue<>(100), // 有界队列
        new ThreadPoolExecutor.CallerRunsPolicy() // 拒绝策略
);
```

使用 Future.get(timeout, unit) 方法设置超时时间：

Future.get(timeout, unit) 方法可以设置等待任务结果的超时时间。如果在指定的时间内任务没有完成，则会抛出 TimeoutException。捕获 TimeoutException 并调用 Future.cancel(true) 方法来取消任务。

ExecutorService executor = Executors.newFixedThreadPool(1);
Future<String> future = executor.submit(new InterruptibleTask());
try {
    String result = future.get(5, TimeUnit.SECONDS); // 设置超时时间为 5 秒
    System.out.println("任务结果：" + result);
} catch (TimeoutException e) {
    System.out.println("任务超时");
    future.cancel(true); // 取消任务
} catch (InterruptedException | ExecutionException e) {
    System.out.println("任务执行出错：" + e.getMessage());
} finally {
    executor.shutdownNow();
}

示例代码：一个完整的超时取消示例

下面是一个完整的示例代码，演示了如何正确地实现任务超时取消。

import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.Socket;
import java.net.SocketTimeoutException;
import java.util.concurrent.*;

public class TimeoutCancellationExample {

    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(1);
        Future<String> future = executor.submit(new Callable<String>() {
            @Override
            public String call() throws Exception {
                try {
                    // 模拟一个阻塞 I/O 操作
                    Socket socket = new Socket("example.com", 80);
                    socket.setSoTimeout(3000); // 设置超时时间为 3 秒
                    BufferedReader reader = new BufferedReader(new InputStreamReader(socket.getInputStream()));
                    String line = reader.readLine();
                    return line;
                } catch (SocketTimeoutException e) {
                    System.out.println("连接超时");
                    return "连接超时";
                } catch (InterruptedException e) {
                    System.out.println("任务被中断");
                    Thread.currentThread().interrupt(); // 重新设置中断状态
                    throw e; // 重新抛出中断异常
                } catch (IOException e) {
                    System.out.println("I/O 异常：" + e.getMessage());
                    throw e;
                }
            }
        });

        try {
            String result = future.get(5, TimeUnit.SECONDS); // 设置超时时间为 5 秒
            System.out.println("任务结果：" + result);
        } catch (TimeoutException e) {
            System.out.println("任务超时");
            future.cancel(true); // 取消任务
        } catch (InterruptedException | ExecutionException e) {
            System.out.println("任务执行出错：" + e.getMessage());
        } finally {
            executor.shutdownNow();
        }
    }
}

在这个示例中，我们创建了一个 Callable 任务，该任务尝试连接到 example.com 的 80 端口并读取一行数据。我们设置了 Socket 的超时时间为 3 秒，并使用 Future.get(5, TimeUnit.SECONDS) 方法设置了等待任务结果的超时时间为 5 秒。

如果在 5 秒内任务没有完成，则会抛出 TimeoutException，我们捕获该异常并调用 future.cancel(true) 方法来取消任务。在 Callable 任务中，我们捕获了 InterruptedException 异常，并重新设置了中断状态，然后重新抛出了该异常。

最后，我们在 finally 块中调用 executor.shutdownNow() 方法来关闭线程池。

总结一些重要的点

总而言之，实现可靠的任务超时取消需要从多个方面入手，包括任务本身的代码、线程池的配置以及异常处理等方面。只有综合考虑这些因素，才能确保任务能够正确地被取消，避免资源泄露和性能问题。

任务代码层面：确保任务能够响应中断信号，并且能够正确地释放资源。
线程池层面：合理配置线程池参数，选择合适的队列类型和拒绝策略。
异常处理层面：正确处理 InterruptedException 异常，并重新抛出中断异常。
超时控制： 使用 Future.get(timeout, unit) 方法设置超时时间，并在超时后调用 Future.cancel(true) 方法取消任务。

希望今天的分享能够帮助大家更好地理解和掌握 Java 并发编程中的任务超时取消技巧。

常见问题及解答

问题	解答
为什么 `future.cancel(true)` 没有效果？	任务可能没有正确响应中断信号，或者任务已经完成/取消/发生异常。
如何确保任务能够响应中断信号？	在任务中周期性地检查中断状态，并使用带有超时时间的 I/O 操作。
`InterruptedException` 应该如何处理？	如果无法完全处理该异常，应该重新抛出该异常，以便让上层调用者知道任务被中断。
线程池应该如何配置？	根据任务的类型和数量，合理设置线程池的核心线程数和最大线程数。使用有界队列可以防止任务过多地堆积在队列中。
如何设置任务的超时时间？	使用 `Future.get(timeout, unit)` 方法可以设置等待任务结果的超时时间。如果在指定的时间内任务没有完成，则会抛出 `TimeoutException`。

希望这些问答能够帮助你更好地理解任务超时取消的机制。