JAVA多线程中FutureTask重复执行与缓存机制原理解析

JAVA多线程中FutureTask重复执行与缓存机制原理解析

大家好，今天我们来深入探讨Java多线程编程中一个非常重要的类：FutureTask。FutureTask不仅可以异步执行任务，还具备缓存和避免重复执行的特性。理解它的工作原理对于编写高效、稳定的并发程序至关重要。

1. FutureTask 概述

FutureTask实现了 RunnableFuture 接口，而 RunnableFuture 接口又继承了 Runnable 和 Future 接口。这意味着 FutureTask 既可以作为一个任务提交给线程池执行，又可以通过 Future 接口获取任务的执行结果。

• Runnable: 允许FutureTask被线程执行。
• Future: 允许获取任务的执行状态和结果。

2. FutureTask 的状态

FutureTask 的生命周期中会经历多种状态，这些状态对于理解其行为至关重要。FutureTask内部通过原子变量 state 来维护这些状态。

状态值	状态名称	描述
NEW	新建状态	FutureTask刚创建，任务尚未开始执行。
COMPLETING	完成中	任务正在执行，并且即将完成，正在设置结果或抛出异常。这是一个短暂的中间状态，用于保证线程安全。
NORMAL	正常完成	任务已经成功执行完成，结果已经设置。可以通过get()方法获取结果。
EXCEPTIONAL	异常完成	任务执行过程中抛出了异常。可以通过get()方法获取异常信息。
CANCELLED	已取消	任务被取消，可能在执行前或执行过程中被取消。
INTERRUPTING	中断中	任务正在尝试中断执行。这也是一个短暂的中间状态。
INTERRUPTED	已中断	任务已经被中断。

3. FutureTask 的核心方法

• FutureTask(Callable<V> callable): 构造函数，接受一个 Callable 对象，该对象代表要执行的任务。
• run(): 执行任务的方法。它会调用 Callable 的 call() 方法，并将结果或异常保存到 FutureTask 中。
• get(): 获取任务的执行结果。如果任务尚未完成，该方法会阻塞，直到任务完成或超时。
• isDone(): 检查任务是否已经完成（正常完成、异常完成或取消）。
• isCancelled(): 检查任务是否已经被取消。
• cancel(boolean mayInterruptIfRunning): 尝试取消任务的执行。mayInterruptIfRunning 参数指定是否允许中断正在运行的任务。

4. FutureTask 的执行流程

1. 创建一个 FutureTask 对象，并将一个 Callable 对象传递给它。
2. 将 FutureTask 对象提交给线程池或直接通过 Thread 启动。
3. FutureTask 的 run() 方法被调用。
4. run() 方法会调用 Callable 的 call() 方法执行任务。
5. 如果 call() 方法成功执行，则将结果保存到 FutureTask 中，并将状态设置为 NORMAL。
6. 如果 call() 方法抛出异常，则将异常保存到 FutureTask 中，并将状态设置为 EXCEPTIONAL。
7. 如果任务被取消，则将状态设置为 CANCELLED。
8. 调用 get() 方法可以获取任务的执行结果。get() 方法会阻塞，直到任务完成或超时。

5. FutureTask 的缓存机制

FutureTask 具有缓存机制，这意味着如果任务已经执行完成，再次调用 get() 方法会立即返回结果，而不会重新执行任务。这是通过 FutureTask 的状态来保证的。只有当 FutureTask 的状态为 NORMAL 或 EXCEPTIONAL 时，get() 方法才会立即返回结果或抛出异常。否则，get() 方法会阻塞，直到任务完成。

6. FutureTask 防止重复执行的机制

FutureTask 的 run() 方法只能被执行一次。这是通过 CAS (Compare and Swap) 操作来保证的。run() 方法首先会尝试将 FutureTask 的状态从 NEW 修改为 COMPLETING 或 INTERRUPTING。如果修改成功，则表示 run() 方法可以执行。如果修改失败，则表示 run() 方法已经被其他线程执行过，或者任务已经被取消，此时 run() 方法会直接返回，不会重复执行任务。

让我们通过代码来更深入地理解 FutureTask 的这些特性。

import java.util.concurrent.*;

public class FutureTaskExample {

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        // 定义一个 Callable 任务
        Callable<String> callable = () -> {
            System.out.println("Executing task...");
            Thread.sleep(2000); // 模拟耗时操作
            return "Task completed!";
        };

        // 创建 FutureTask 对象
        FutureTask<String> futureTask = new FutureTask<>(callable);

        // 创建线程池
        ExecutorService executor = Executors.newFixedThreadPool(2);

        // 提交 FutureTask 到线程池
        executor.submit(futureTask);

        // 第一次获取结果
        System.out.println("Getting result for the first time...");
        String result1 = futureTask.get(); // 阻塞直到任务完成
        System.out.println("Result 1: " + result1);

        // 第二次获取结果
        System.out.println("Getting result for the second time...");
        String result2 = futureTask.get(); // 立即返回结果，不会重新执行任务
        System.out.println("Result 2: " + result2);

        // 检查任务是否完成
        System.out.println("Is task done? " + futureTask.isDone());

        // 关闭线程池
        executor.shutdown();
        executor.awaitTermination(1, TimeUnit.MINUTES);
    }
}

在这个例子中，我们创建了一个 Callable 任务，它会模拟一个耗时操作。然后，我们创建了一个 FutureTask 对象，并将 Callable 对象传递给它。我们将 FutureTask 提交给线程池执行。第一次调用 get() 方法时，由于任务尚未完成，所以 get() 方法会阻塞，直到任务完成。第二次调用 get() 方法时，由于任务已经完成，所以 get() 方法会立即返回结果，而不会重新执行任务。这就是 FutureTask 的缓存机制。

下面我们来演示 FutureTask 如何防止重复执行。

import java.util.concurrent.*;

public class FutureTaskDuplicateExecution {

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        Callable<String> callable = () -> {
            System.out.println("Executing task...");
            Thread.sleep(1000);
            return "Task completed!";
        };

        FutureTask<String> futureTask = new FutureTask<>(callable);

        // 创建两个线程来执行同一个 FutureTask
        Thread thread1 = new Thread(futureTask);
        Thread thread2 = new Thread(futureTask);

        thread1.start();
        thread2.start(); // 只有一个线程会成功执行 run() 方法

        thread1.join();
        thread2.join();

        System.out.println("Task is done: " + futureTask.isDone());
        System.out.println("Result: " + futureTask.get());
    }
}

在这个例子中，我们创建了两个线程来执行同一个 FutureTask 对象。只有一个线程会成功执行 run() 方法，另一个线程会因为 CAS 操作失败而直接返回。这就是 FutureTask 防止重复执行的机制。

7. 异常处理

FutureTask 可以捕获任务执行过程中抛出的异常。如果 Callable 的 call() 方法抛出异常，则 FutureTask 会将异常保存起来，并在调用 get() 方法时抛出 ExecutionException 异常。

import java.util.concurrent.*;

public class FutureTaskExceptionHandling {

    public static void main(String[] args) throws InterruptedException {
        Callable<String> callable = () -> {
            System.out.println("Executing task...");
            throw new RuntimeException("Task failed!");
        };

        FutureTask<String> futureTask = new FutureTask<>(callable);

        ExecutorService executor = Executors.newFixedThreadPool(1);
        executor.submit(futureTask);

        try {
            futureTask.get(); // 获取结果，会抛出 ExecutionException
        } catch (ExecutionException e) {
            System.out.println("Exception caught: " + e.getCause()); // 获取原始异常
        }

        executor.shutdown();
        executor.awaitTermination(1, TimeUnit.MINUTES);
    }
}

在这个例子中，Callable 的 call() 方法会抛出一个 RuntimeException 异常。当调用 get() 方法时，会抛出一个 ExecutionException 异常，我们可以通过 getCause() 方法获取原始的 RuntimeException 异常。

8. 取消任务

FutureTask 允许取消任务的执行。可以通过调用 cancel() 方法来取消任务。cancel() 方法接受一个 mayInterruptIfRunning 参数，该参数指定是否允许中断正在运行的任务。

• 如果 mayInterruptIfRunning 为 true，则 cancel() 方法会尝试中断正在运行的任务。
• 如果 mayInterruptIfRunning 为 false，则 cancel() 方法只会阻止任务在未来执行。

import java.util.concurrent.*;

public class FutureTaskCancellation {

    public static void main(String[] args) throws InterruptedException, ExecutionException {
        Callable<String> callable = () -> {
            System.out.println("Executing task...");
            try {
                Thread.sleep(5000); // 模拟耗时操作
            } catch (InterruptedException e) {
                System.out.println("Task interrupted!");
                return "Task interrupted";
            }
            return "Task completed!";
        };

        FutureTask<String> futureTask = new FutureTask<>(callable);

        ExecutorService executor = Executors.newFixedThreadPool(1);
        executor.submit(futureTask);

        Thread.sleep(1000); // 等待任务执行一段时间

        boolean cancelled = futureTask.cancel(true); // 尝试中断任务
        System.out.println("Task cancelled: " + cancelled);

        try {
            String result = futureTask.get(); // 获取结果，会抛出 CancellationException
            System.out.println("Result: " + result);
        } catch (CancellationException e) {
            System.out.println("Task was cancelled.");
        }

        executor.shutdown();
        executor.awaitTermination(1, TimeUnit.MINUTES);
    }
}

在这个例子中，我们首先提交了一个耗时任务到线程池。然后，我们等待一段时间，并尝试取消任务。如果 cancel() 方法返回 true，则表示任务已经被成功取消。当调用 get() 方法时，会抛出一个 CancellationException 异常。

9. FutureTask 的应用场景

• 异步计算: FutureTask 非常适合执行需要花费较长时间的异步计算任务，例如网络请求、数据库查询等。
• 缓存: FutureTask 可以作为缓存使用，避免重复执行相同的计算任务。
• 任务依赖: FutureTask 可以用于构建复杂的任务依赖关系，例如先执行任务 A，再执行任务 B，最后执行任务 C。
• 并发控制: FutureTask 可以用于控制并发执行的任务数量。

10. FutureTask 的注意事项

• 避免在 get() 方法中无限期阻塞。应该设置超时时间，以防止程序卡死。
• 小心处理 ExecutionException 和 CancellationException 异常。
• 理解 cancel() 方法的 mayInterruptIfRunning 参数的含义。
• 合理使用线程池，避免创建过多的线程。

掌握FutureTask，提升并发编程能力

FutureTask 是 Java 并发编程中一个非常重要的工具类。它提供了异步执行任务、缓存结果、防止重复执行、异常处理和取消任务等功能。掌握 FutureTask 的工作原理对于编写高效、稳定的并发程序至关重要。

关于重复执行和结果缓存

通过状态管理和CAS操作，FutureTask确保任务只被执行一次，并将结果缓存起来，避免重复计算。

异常处理和任务取消机制

FutureTask能够捕获任务执行中的异常，并在获取结果时抛出。同时，它提供了取消任务的功能，可以中断正在执行的任务或阻止任务的执行。