掌握Java并发工具：Callable和Future实战技巧

Callable接口介绍

Callable vs Runnable

在Java中，Callable接口是一个返回结果并可能抛出异常的任务。它类似于Runnable接口，但有两个显著的不同：

Callable的call()方法可以返回值。
call()方法可以抛出受检查的异常。

java">import java.util.concurrent.Callable;public class WordCallable implements Callable<String> {@Overridepublic String call() throws Exception {// 休眠一秒模拟长时间运行的任务Thread.sleep(1000);return "Hello, Callable!";}
}

Callable的优势

相比于Runnable，Callable最大的优势在于它的灵活性：

它可以直接通过Future得到任务执行的结果。
能够处理更复杂的业务逻辑，因为它可以抛出异常。
它广泛应用在需要任务执行结果时。

java">import java.util.concurrent.FutureTask;public class CallableExample {public static void main(String args) throws Exception {WordCallable wordCallable = new WordCallable();FutureTask<String> futureTask = new FutureTask<>(wordCallable);new Thread(futureTask).start();// 获取异步执行的结果String result = futureTask.get();System.out.println(result);}
}

实现Callable接口的重要类分析

Executors类

Executors类是java.util.concurrent包中的一个工厂和工具类，用于创建线程池和Future对象。它提供了多种静态方法来创建不同类型的线程池，例如newFixedThreadPool、newCachedThreadPool等。

java">import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;public class CallableWithExecutors {public static void main(String args) throws Exception {ExecutorService executor = Executors.newSingleThreadExecutor();Callable<String> callable = new WordCallable();Future<String> future = executor.submit(callable);// 做一些其他的事情...String result = future.get();System.out.println(result);executor.shutdown();}
}

FutureTask类

FutureTask是Future的一个具体实现，它同样实现了Runnable，因此它既可以由Thread对象执行，也可以提交给ExecutorService。

java">import java.util.concurrent.FutureTask;public class FutureTaskExample {public static void main(String args) {Callable<String> callable = new WordCallable();FutureTask<String> task = new FutureTask<>(callable);new Thread(task).start();// 当需要结果时获取，可能会阻塞String result = task.get();System.out.println(result);}
}

两种异步模型与深度解析

Future接口

Future接口功能

Future接口在Java并发中是用来描述一个异步计算的结果。通过Future对象，我们可以了解任务执行情况，取消任务，以及获取执行结果。

java">import java.util.concurrent.*;public class SimpleFutureDemo {public static void main(String[] args) throws ExecutionException, InterruptedException {ExecutorService executorService = Executors.newFixedThreadPool(1);Future<Integer> futureResult = executorService.submit(() -> {TimeUnit.SECONDS.sleep(2);return 42;});// 任务进行中的时候可以做一些其他事情// ...// 当我们需要这个结果的时候，我们可以调用 get// 如果任务已完成，get会立即返回结果，否则会阻塞直到任务进入完成状态，然后返回结果或者抛出异常Integer result = futureResult.get();System.out.println("Future result is: " + result);executorService.shutdown();}
}

Future接口的方法解读

Future接口提供以下几个关键方法：

cancel：试图取消运行中的任务
isCancelled：返回布尔值，如果任务在正常完成前被取消，则为true
isDone：如果任务已完成，则返回true
get：等待计算完成，然后检索其结果
get (with timeout)：如果计算已完成，则检索其结果，否则等待指定的时间

java">// 省略之前Future接口功能的代码示例...
Integer result = null;
try {// 在指定的时间内等待结果，如果任务在指定时间内完成，将返回结果，否则抛出TimeoutExceptionresult = futureResult.get(1, TimeUnit.SECONDS);
} catch (TimeoutException e) {System.err.println("任务超时未获取到结果");
} catch (ExecutionException e) {System.err.println("任务执行异常");
} catch (InterruptedException e) {System.err.println("任务被中断");
}
System.out.println("Future result is: " + result);
// 关闭线程池
executorService.shutdown();

两种异步模型

file

Executor框架模型

Executor框架是Java提供的一种用来管理线程资源的框架，通过使用它可以简化多线程编程的复杂性。Executor框架主要由Executor, Executors, ExecutorService, CompletionService, ThreadPoolExecutor等组成。

java">// 示例中我们之前已经使用到了ExecutorService和Executors
// 假设我们有更多的Callable任务需要执行，我们可以使用线程池来管理这些任务

Fork/Join框架模型

Fork/Join框架主要用于并行执行任务和处理大量数据。这个模型的思想是将一个大任务分割成若干小任务，这些小任务分别执行，最后将小任务的结果合并得到大任务结果的模式。

java">import java.util.concurrent.RecursiveTask;public class FibonacciComputation extends RecursiveTask<Integer> {final int n;FibonacciComputation(int n) {this.n = n;}@Overrideprotected Integer compute() {if (n <= 1)return n;FibonacciComputation f1 = new FibonacciComputation(n - 1);f1.fork(); // 开始异步执行FibonacciComputation f2 = new FibonacciComputation(n - 2);return f2.compute() + f1.join(); // 等待异步执行结果和当前结果相加}
}

深度解析Future接口

如何取消任务

Future接口的cancel方法可以用来取消任务。如果取消成功，以后尝试通过get方法获取结果时会抛出CancellationException。

java">// 省略之前Future接口功能的代码示例...// 取消任务
boolean cancelled = futureResult.cancel(true);
// 验证任务是否已经被取消
System.out.println("Task was cancelled: " + cancelled);

如何检查任务状态

可以通过调用isDone和isCancelled方法来检查任务的状态。

java">// 省略之前Future接口功能的代码示例...// 检查任务是否已经取消
System.out.println("Task was cancelled: " + futureResult.isCancelled());
// 检查任务是否已完成
System.out.println("Task is done: " + futureResult.isDone());

处理超时

使用带有超时的get方法来处理长时间未完成的任务可能导致的等待问题。

java">// 省略之前Future接口功能的代码示例...
// 使用get(long timeout, TimeUnit unit)方法，可以防止无限期等待任务的完成。

Future接口的局限性和解决方案

尽管Future提供了对于异步任务的基本控制，它有一定的局限性，比如无法直接得知任务完成的进度，不支持完成通知，也不支持链式Future操作。Java 8引入了CompletableFuture，它提供了更多的灵活性和控制力，包括任务结果合成、异常处理、事件完成通知等。

java">import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;public class CompletableFutureDemo {public static void main(String[] args) throws ExecutionException, InterruptedException {CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello CompletableFuture!");// 当异步任务完成或者发生异常时，可以手动完成或者处理异常future.complete("手动完成的结果");future.exceptionally(ex -> "出现异常: " + ex.getMessage());System.out.println("CompletableFuture result: " + future.get());}
}