Future是JDK1.5 提供的接口，是用来以阻塞的方式获取线程异步执行完的结果。
FutureTask 类是 Java 中 Future 接口的一个实现，同时也实现了 Runnable 接口。它用于表示异步计算的结果，允许一个任务在一个线程中计算结果，在另一个线程中获取计算的结果。
import java.util.concurrent.*;
public class CallableWithThreadPoolExample {
public static void main(String[] args) {
// 创建一个固定大小的线程池
ExecutorService executorService = Executors.newFixedThreadPool(2);
// 创建一个Callable任务
Callable<Integer> callableTask = () -> {
// 模拟耗时的计算
try {
Thread.sleep(2000);
} catch (InterruptedException e) {
e.printStackTrace();
}
// 返回计算结果
return 42;
};
// 提交Callable任务给线程池执行
Future<Integer> future = executorService.submit(callableTask);
// 执行其他任务，不会阻塞
try {
// 获取Callable任务的结果，会阻塞直到结果准备好
Integer result = future.get();
System.out.println("Result: " + result);
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
} finally {
// 关闭线程池
executorService.shutdown();
}
}
}
FutureTask 提供了 get 方法，可以用于获取异步计算的结果。然而，FutureTask 本身并没有保证按照任务提交的顺序返回结果。
如果你需要按照任务提交的顺序获取执行结果，你可以使用 ExecutorService 的 invokeAll 方法提交一批 Callable 任务，并得到一组 Future 对象，然后可以通过迭代这组 Future 对象来获取执行结果，迭代的顺序即为任务提交的顺序。
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.*;
public class OrderPreservingExample {
public static void main(String[] args) {
ExecutorService executorService = Executors.newFixedThreadPool(5);
List<Callable<String>> tasks = new ArrayList<>();
for (int i = 0; i < 10; i++) {
final int taskId = i;
Callable<String> task = () -> {
// Simulate some computation
Thread.sleep(1000);
return "Task " + taskId + " completed";
};
tasks.add(task);
}
try {
List<Future<String>> results = executorService.invokeAll(tasks);
for (Future<String> result : results) {
System.out.println(result.get()); // Results are obtained in the order of task submission
}
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
} finally {
executorService.shutdown();
}
}
}
在上面的示例中，invokeAll 方法会按照任务列表的顺序返回 Future 对象的列表，因此通过迭代这个列表，可以按照任务提交的顺序获取执行结果。请注意，invokeAll 方法会阻塞直到所有任务完成。
CompletionService 是 Java 中用于处理一批异步任务的工具类，它允许你以异步的方式提交任务，并在任务完成时按照完成顺序获取结果。CompletionService 的底层原理主要基于阻塞队列和线程池。
CompletionService 使用一个阻塞队列来保存已完成的任务。当一个任务完成时，它会被放入队列中。阻塞队列的选择通常是 LinkedBlockingQueue，它是一个先进先出的队列，确保按照任务完成的顺序排列。
CompletionService 通常与 Executor 框架一起使用。创建一个 ExecutorService，并将其传递给 CompletionService 的构造函数。这个线程池负责执行提交的任务。
当想要获取已完成的任务的结果时，可以调用 CompletionService 的 take() 或 poll() 方法。这些方法会从阻塞队列中取出已完成的任务的 Future，并返回它。如果队列为空，take() 方法会阻塞，而 poll() 方法会返回 null。
由于使用了阻塞队列，你可以确保按照任务完成的顺序获取结果，即使任务的完成顺序与它们被提交的顺序不同。
Future 接口本身并没有直接提供按照任务提交的顺序获取执行结果的机制。当通过 ExecutorService 提交多个任务时，Future 对象的完成顺序可能不一定与任务的提交顺序完全一致。
ExecutorCompletionService 是 ExecutorService 的一个实现，它可以将已完成的任务放入一个阻塞队列中，然后可以通过检索队列的元素来按照任务完成的顺序获取结果。
当向 CompletionService 提交一个任务时，它会将该任务包装在一个 Future 中，并将这个 Future 放入阻塞队列中。
import java.util.concurrent.*;
public class OrderPreservingFutureExample {
public static void main(String[] args) {
ExecutorService executorService = Executors.newFixedThreadPool(5);
ExecutorCompletionService<String> completionService = new ExecutorCompletionService<>(executorService);
for (int i = 0; i < 10; i++) {
final int taskId = i;
Callable<String> task = () -> {
// Simulate some computation
Thread.sleep(1000);
return "Task " + taskId + " completed";
};
completionService.submit(task);
}
try {
for (int i = 0; i < 10; i++) {
Future<String> result = completionService.take(); // Blocking until a task is completed
System.out.println(result.get()); // Results are obtained in the order of completion
}
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
} finally {
executorService.shutdown();
}
}
}
在上面的示例中，ExecutorCompletionService 的 take 方法会阻塞直到有任务完成，然后返回一个 Future 对象，我们能够通过这个对象获取任务的执行结果。通过循环迭代 take 方法，可以按照任务完成的顺序获取执行结果。
这是因为CompletionService 的 take 方法是阻塞的。当调用 take 方法时，它会等待至少一个任务完成并返回一个包含已完成任务结果的 Future 对象。如果当前没有已完成的任务，take 方法会阻塞直到有任务完成为止。这种阻塞行为使得能够按照任务完成的顺序获取结果，因为它会返回最先完成的任务的结果。
如果我们希望非阻塞地检查是否有任务完成，可以使用 poll 方法，它会立即返回一个 Future 对象，如果没有已完成的任务，则返回 null。但使用 poll 方法可能需要在循环中进行主动轮询。
import java.util.concurrent.*;
public class PollingCompletionServiceExample {
public static void main(String[] args) {
ExecutorService executorService = Executors.newFixedThreadPool(5);
CompletionService<String> completionService = new ExecutorCompletionService<>(executorService);
// 提交任务到CompletionService
for (int i = 0; i < 5; i++) {
final int taskId = i;
Callable<String> task = () -> {
// 模拟一些计算
Thread.sleep(1000);
return "Task " + taskId + " completed";
};
completionService.submit(task);
}
// 轮询任务的完成状态
for (int i = 0; i < 5; i++) {
Future<String> result = completionService.poll();
if (result != null) {
// 任务已完成，处理结果
try {
System.out.println(result.get());
} catch (InterruptedException | ExecutionException e) {
e.printStackTrace();
}
} else {
// 没有已完成的任务
System.out.println("No completed tasks at the moment.");
}
}
executorService.shutdown();
}
}
在上面的示例中，poll方法被用于轮询已完成的任务。如果有任务已经完成，它会返回一个包含任务结果的Future对象；否则，返回null。需要注意的是，如果任务尚未完成，poll方法会立即返回null，因此你可能需要在循环中添加适当的延迟，以避免过于频繁地检查任务状态。
Future和CompletionService 在实际使用过程中存在一些局限性比如不支持异步任务的编排组合、获取计算结果的 get() /take()为阻塞调用。
Java 8 才被引入CompletableFuture 类可以解决Future 的这些缺陷。CompletableFuture 除了提供了更为好用和强大的 Future 特性之外，还提供了函数式编程、异步任务编排组合（可以将多个异步任务串联起来，组成一个完整的链式调用）等能力。
CompletableFuture 同时实现了 Future 和 CompletionStage 接口。
CompletionStage 接口描述了一个异步计算的阶段。很多计算可以分成多个阶段或步骤，此时可以通过它将所有步骤组合起来，形成异步计算的流水线。
CompletableFuture 除了提供了更为好用和强大的 Future 特性之外，还提供了函数式编程的能力。
通过 thenApply, thenAccept, 或者 thenRun 方法，注册回调函数，这些函数会在 CompletableFuture 完成时被异步调用。这样，处理任务的结果而不必阻塞当前线程。
CompletableFuture<String> future = CompletableFuture.supplyAsync(() -> "Hello");
future.thenApply(result -> {
// Non-blocking callback to process the result
System.out.println("Received result: " + result);
return result.toUpperCase();
});
// Continue with other non-blocking operations
使用 thenCombine, thenAcceptBoth, runAfterBoth, applyToEither, acceptEither, 等方法，将多个 CompletableFuture 的结果组合在一起，而不必阻塞等待每个任务的完成。
CompletableFuture<String> firstTask = CompletableFuture.supplyAsync(() -> {
// Simulate some computation
return "First Task";
});
CompletableFuture<String> secondTask = CompletableFuture.supplyAsync(() -> {
// Simulate some computation
return "Second Task";
});
CompletableFuture<String> thirdTask = CompletableFuture.supplyAsync(() -> {
// Simulate some computation
return "Third Task";
});
// 使用thenCompose确保任务按照顺序完成
CompletableFuture<String> result = firstTask.thenCompose(result1 ->
secondTask.thenCompose(result2 ->
thirdTask.thenApply(result3 -> result1 + " -> " + result2 + " -> " + result3)
)
);
// 异步获取结果
result.thenAcceptAsync(System.out::println);
// 阻塞等待所有任务完成
result.join();