深入理解高并发编程 - Callable 和 Future

ゝ一纸荒年。 2023-10-14 12:39 34阅读 0赞

在 Java 中，Callable 和 Future 是用于处理多线程编程的两个关键接口，它们位于 java.util.concurrent 包下。

## 1、Callable： ##

Callable 接口是一个泛型接口，它定义了一个带有返回值的任务。与 Runnable 接口不同，Callable 的 call 方法可以返回一个值，并且可以抛出异常。Callable 接口只有一个方法：`V call() throws Exception;`，其中 V 是表示返回值的类型。

例如，可以创建一个计算斐波那契数列的 Callable 对象：

import java.util.concurrent.Callable;
    
    public class FibonacciCallable implements Callable<Integer> {
        
        private int n;
    
        public FibonacciCallable(int n) {
        
            this.n = n;
        }
    
        @Override
        public Integer call() throws Exception {
        
            return fibonacci(n);
        }
    
        private int fibonacci(int n) {
        
            if (n <= 1) {
        
                return n;
            }
            return fibonacci(n - 1) + fibonacci(n - 2);
        }
    }

### 1.1、有哪些类实现了 Callable 接口 ###

在 Java 标准库中，有一些类实现了 Callable 接口，允许创建并执行具有返回值的异步任务。以下是一些常见的实现了 Callable 接口的类：

#### 1.1.1、FutureTask： ####

FutureTask 是一个实现了 RunnableFuture 接口的类，而 RunnableFuture 扩展了 Runnable 和 Future 接口。它可以包装一个 Callable 或 Runnable 对象，允许你在单独的线程中执行该任务，并在之后获取任务的结果。这是一个常用的工具类，可用于处理异步任务。

import java.util.concurrent.Callable;
    import java.util.concurrent.ExecutionException;
    import java.util.concurrent.FutureTask;
    
    public class MyFutureTaskExample {
        
        public static void main(String[] args) {
        
            // 创建一个 Callable 任务
            Callable<Integer> callable = new MyFibonacciCallable(10);
    
            // 将 Callable 包装成 FutureTask
            FutureTask<Integer> futureTask = new FutureTask<>(callable);
    
            // 创建线程并执行 FutureTask
            Thread thread = new Thread(futureTask);
            thread.start();
    
            try {
        
                // 等待任务完成并获取结果
                int result = futureTask.get();
                System.out.println("Fibonacci result: " + result);
            } catch (InterruptedException | ExecutionException e) {
        
                e.printStackTrace();
            }
        }
    }

#### 1.1.2、ThreadPoolExecutor： ####

ThreadPoolExecutor 是 Java 标准库中用于创建和管理线程池的类。它可以执行 Runnable 或 Callable 任务，并管理线程池的大小、线程复用等。当提交一个 Callable 任务时，它会返回一个 Future 对象，通过该对象可以获取任务的结果。

import java.util.concurrent.*;
    
    public class MyThreadPoolExecutorExample {
        
        public static void main(String[] args) {
        
            // 创建 ThreadPoolExecutor，设置核心线程数、最大线程数、队列等待数等参数
            ThreadPoolExecutor executor = new ThreadPoolExecutor(
                    2, // 核心线程数
                    4, // 最大线程数
                    60, // 空闲线程存活时间
                    TimeUnit.SECONDS, // 存活时间单位
                    new LinkedBlockingQueue<>() // 任务队列
            );
    
            // 创建一个 Callable 任务
            Callable<Integer> callable = new MyFibonacciCallable(10);
    
            try {
        
                // 提交 Callable 任务并获取 Future 对象
                Future<Integer> future = executor.submit(callable);
    
                // 等待任务完成并获取结果
                int result = future.get();
                System.out.println("Fibonacci result: " + result);
            } catch (InterruptedException | ExecutionException e) {
        
                e.printStackTrace();
            }
    
            // 关闭线程池
            executor.shutdown();
        }
    }

#### 1.1.3、ScheduledThreadPoolExecutor： ####

这是一个继承自 ThreadPoolExecutor 的类，用于在指定的时间或周期性地执行任务。它可以执行 Callable 任务，并通过 Future 获取结果。

import java.util.concurrent.*;
    
    public class MyScheduledThreadPoolExecutorExample {
        
        public static void main(String[] args) {
        
            // 创建 ScheduledThreadPoolExecutor，设置核心线程数
            ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1);
    
            // 创建一个 Callable 任务
            Callable<Integer> callable = new MyFibonacciCallable(10);
    
            // 延迟 2 秒后执行任务，并返回一个 Future 对象
            ScheduledFuture<Integer> future = executor.schedule(callable, 2, TimeUnit.SECONDS);
    
            try {
        
                // 等待任务完成并获取结果
                int result = future.get();
                System.out.println("Fibonacci result: " + result);
            } catch (InterruptedException | ExecutionException e) {
        
                e.printStackTrace();
            }
    
            // 关闭线程池
            executor.shutdown();
        }
    }

#### 1.1.4、ExecutorService 的其他实现类： ####

除了上述提到的 ThreadPoolExecutor 和 ScheduledThreadPoolExecutor，ExecutorService 接口的其他实现类（如 FixedThreadPool、CachedThreadPool 等）也可以执行 Callable 任务并返回 Future 对象。

import java.util.concurrent.*;
    
    public class MyExecutorServiceExample {
        
        public static void main(String[] args) {
        
            // 创建一个固定大小的线程池
            ExecutorService executor = Executors.newFixedThreadPool(2);
    
            // 创建一个 Callable 任务
            Callable<Integer> callable = new MyFibonacciCallable(10);
    
            try {
        
                // 提交 Callable 任务并获取 Future 对象
                Future<Integer> future = executor.submit(callable);
    
                // 等待任务完成并获取结果
                int result = future.get();
                System.out.println("Fibonacci result: " + result);
            } catch (InterruptedException | ExecutionException e) {
        
                e.printStackTrace();
            }
    
            // 关闭线程池
            executor.shutdown();
        }
    }

#### 1.1.5、ParallelStream： ####

Java 8 引入的 Stream API 中的 parallelStream() 方法可以将集合转换成并行流，能够并行地执行任务。可以在并行流中执行 Callable 任务，并通过 Collectors.toList() 等方法收集任务的结果。

import java.util.List;
    import java.util.stream.Collectors;
    import java.util.stream.IntStream;
    
    public class MyParallelStreamExample {
        
        public static void main(String[] args) {
        
            // 创建一个列表
            List<Integer> numbers = IntStream.rangeClosed(1, 10).boxed().collect(Collectors.toList());
    
            // 并行流处理
            // 模拟执行 Callable 任务
            List<Integer> results = numbers.parallelStream()
                    .map(MyParallelStreamExample::simulateCallableTask)
                    .collect(Collectors.toList());
    
            System.out.println("Results: " + results);
        }
    
        private static int simulateCallableTask(int number) {
        
            // 模拟计算
            return number * number;
        }
    }

## 2、Future： ##

Future 接口表示一个可能还没有完成的异步任务的结果。当提交一个 Callable 对象给一个线程池执行时，会得到一个 Future 对象，通过这个对象可以获取任务的执行结果。

可以使用 Future 的方法来获取任务的执行结果，或者等待任务完成。以下是一些常用的方法：

V get() throws InterruptedException, ExecutionException：获取任务的结果，如果任务尚未完成，这个方法会阻塞直到任务完成。
    V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException：与上面的方法类似，但是可以设置最大等待时间。
    boolean isDone()：判断任务是否已经完成。
    boolean isCancelled()：判断任务是否被取消。
    boolean cancel(boolean mayInterruptIfRunning)：尝试取消任务的执行。

示例使用 Future 来获取 Callable 的执行结果：

import java.util.concurrent.*;
    
    public class FutureExample {
        
        public static void main(String[] args) throws InterruptedException, ExecutionException {
        
            ExecutorService executor = Executors.newSingleThreadExecutor();
            Callable<Integer> callable = new FibonacciCallable(10);
            Future<Integer> future = executor.submit(callable);
    
            // 获取任务结果
            int result = future.get();
            System.out.println("Fibonacci result: " + result);
    
            executor.shutdown();
        }
    }

### 2.1 Future 各个方法介绍 ###

#### 2.1.1、V get() throws InterruptedException, ExecutionException： ####

V get() throws InterruptedException, ExecutionException 方法用于获取任务的结果。如果任务已经完成，它会立即返回结果；如果任务尚未完成，它会阻塞当前线程直到任务完成。如果任务抛出异常，调用 get() 方法会抛出 ExecutionException，其中的 getCause() 方法可以获取实际的异常。

import java.util.concurrent.*;
    
    public class MyFutureGetExample {
        
        public static void main(String[] args) {
        
            // 创建线程池
            ExecutorService executor = Executors.newSingleThreadExecutor();
    
            // 创建一个 Callable 任务
            Callable<Integer> callable = () -> {
        
                // 模拟耗时计算
                Thread.sleep(2000);
                return 42;
            };
    
            // 提交 Callable 任务并获取 Future 对象
            Future<Integer> future = executor.submit(callable);
    
            try {
        
                System.out.println("Waiting for the result...");
    
                // 获取任务的结果，阻塞直到任务完成
                int result = future.get();
    
                System.out.println("Result: " + result);
            } catch (InterruptedException e) {
        
                e.printStackTrace();
            } catch (ExecutionException e) {
        
                Throwable cause = e.getCause(); // 获取实际的异常
                System.err.println("An exception occurred: " + cause.getMessage());
            }
    
            // 关闭线程池
            executor.shutdown();
        }
    }

#### 2.1.2、V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException： ####

V get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException 方法与 get() 方法类似，但是可以设置最大等待时间。如果任务在指定时间内完成，将返回结果；如果超时，将抛出 TimeoutException。

import java.util.concurrent.*;
    
    public class MyFutureGetWithTimeoutExample {
        
        public static void main(String[] args) {
        
            // 创建线程池
            ExecutorService executor = Executors.newSingleThreadExecutor();
    
            // 创建一个 Callable 任务
            Callable<Integer> callable = () -> {
        
                // 模拟耗时计算
                Thread.sleep(4000);
                // Thread.sleep(2000);
                return 42;
            };
    
            // 提交 Callable 任务并获取 Future 对象
            Future<Integer> future = executor.submit(callable);
    
            try {
        
                System.out.println("Waiting for the result...");
    
                // 设置最大等待时间为 3 秒
                int timeout = 3;
                TimeUnit timeUnit = TimeUnit.SECONDS;
    
                // 获取任务的结果，等待最多 3 秒，超时后抛出 TimeoutException
                int result = future.get(timeout, timeUnit);
    
                System.out.println("Result: " + result);
            } catch (InterruptedException e) {
        
                e.printStackTrace();
            } catch (ExecutionException e) {
        
                Throwable cause = e.getCause(); // 获取实际的异常
                System.err.println("An exception occurred: " + cause.getMessage());
            } catch (TimeoutException e) {
        
                System.err.println("Timeout occurred while waiting for the result.");
            }
    
            // 关闭线程池
            executor.shutdown();
        }
    }

#### 2.1.3、boolean isDone()： ####

boolean isDone() 方法用于判断任务是否已经完成。如果任务已完成（无论是正常完成还是异常完成），则返回 true，否则返回 false。

import java.util.concurrent.*;
    
    public class MyFutureIsDoneExample {
        
        public static void main(String[] args) {
        
            // 创建线程池
            ExecutorService executor = Executors.newSingleThreadExecutor();
    
            // 创建一个 Callable 任务
            Callable<Integer> callable = () -> {
        
                // 模拟耗时计算
                Thread.sleep(2000);
                return 42;
            };
    
            // 提交 Callable 任务并获取 Future 对象
            Future<Integer> future = executor.submit(callable);
    
            System.out.println("Waiting for the result...");
    
            // 判断任务是否已完成
            while (!future.isDone()) {
        
                try {
        
                    // 每隔 500 毫秒检查一次任务状态
                    Thread.sleep(500);
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
            }
    
            try {
        
                // 获取任务的结果
                int result = future.get();
                System.out.println("Result: " + result);
            } catch (InterruptedException e) {
        
                e.printStackTrace();
            } catch (ExecutionException e) {
        
                Throwable cause = e.getCause(); // 获取实际的异常
                System.err.println("An exception occurred: " + cause.getMessage());
            }
    
            // 关闭线程池
            executor.shutdown();
        }
    }

#### 2.1.4、boolean isCancelled()： ####

boolean isCancelled() 方法用于判断任务是否被取消。如果任务在完成之前被取消，返回 true，否则返回 false。

import java.util.concurrent.*;
    
    public class MyFutureIsCancelledExample {
        
        public static void main(String[] args) {
        
            // 创建线程池
            ExecutorService executor = Executors.newSingleThreadExecutor();
    
            // 创建一个 Callable 任务
            Callable<Integer> callable = () -> {
        
                // 模拟耗时计算
                Thread.sleep(2000);
                return 42;
            };
    
            // 提交 Callable 任务并获取 Future 对象
            Future<Integer> future = executor.submit(callable);
    
            // 尝试取消任务的执行
            boolean cancelled = future.cancel(true);
    
            if (cancelled) {
        
                System.out.println("Task has been cancelled.");
            } else {
        
                System.out.println("Task cannot be cancelled or is already completed.");
            }
    
            // 判断任务是否被取消
            if (future.isCancelled()) {
        
                System.out.println("Task has been cancelled.");
            } else {
        
                System.out.println("Task is not cancelled.");
            }
    
            // 关闭线程池
            executor.shutdown();
        }
    }

#### 2.1.5、boolean cancel(boolean mayInterruptIfRunning)： ####

boolean cancel(boolean mayInterruptIfRunning) 方法用于尝试取消任务的执行。如果任务尚未开始执行，或者已经执行完毕，或者参数 mayInterruptIfRunning 为 false，则任务不会被取消并返回 false。如果任务正在执行或尚未开始执行，并且参数 mayInterruptIfRunning 为 true，则任务会被中断（如果支持中断），并返回 true。

import java.util.concurrent.*;
    
    public class MyFutureCancelExample {
        
        public static void main(String[] args) {
        
            // 创建线程池
            ExecutorService executor = Executors.newSingleThreadExecutor();
    
            // 创建一个 Callable 任务
            Callable<Integer> callable = () -> {
        
                System.out.println("Task started.");
                // 模拟耗时计算
                Thread.sleep(5000);
                System.out.println("Task completed.");
                return 42;
            };
    
            // 提交 Callable 任务并获取 Future 对象
            Future<Integer> future = executor.submit(callable);
    
            // 尝试取消任务的执行
            boolean cancelled = future.cancel(true);
    
            if (cancelled) {
        
                System.out.println("Task has been cancelled.");
            } else {
        
                System.out.println("Task cannot be cancelled or is already completed.");
            }
    
            // 关闭线程池
            executor.shutdown();
        }
    }

### 2.2、有哪些类实现了 Future接口 ###

在 Java 标准库中，以下是一些实现了 Future 接口的主要类：

#### 2.2.1、FutureTask： ####

FutureTask 是一个实现了 RunnableFuture 接口的类，同时也是 Future 的实现类。它可以包装一个 Callable 或 Runnable 对象，允许在单独的线程中执行任务，并在之后获取任务的结果。它通常与 ExecutorService 配合使用，也可以直接通过 Thread 来执行。除了实现 Future 接口，FutureTask 还实现了 Runnable 接口，可以作为线程的任务进行执行。

import java.util.concurrent.*;
    
    public class MyExecutorServiceFutureTaskExample {
        
        public static void main(String[] args) {
        
            // 创建线程池
            ExecutorService executor = Executors.newSingleThreadExecutor();
    
            // 创建一个 Callable 任务
            Callable<Integer> callable = () -> {
        
                // 模拟耗时计算
                Thread.sleep(2000);
                return 42;
            };
    
            // 创建 FutureTask 对象
            FutureTask<Integer> futureTask = new FutureTask<>(callable);
    
            // 提交 FutureTask 给线程池执行
            executor.submit(futureTask);
    
            System.out.println("Task submitted. Waiting for the result...");
    
            try {
        
                // 获取任务的结果，如果任务尚未完成，会阻塞直到任务完成
                int result = futureTask.get();
                System.out.println("Result: " + result);
            } catch (InterruptedException e) {
        
                e.printStackTrace();
            } catch (ExecutionException e) {
        
                Throwable cause = e.getCause(); // 获取实际的异常
                System.err.println("An exception occurred: " + cause.getMessage());
            }
    
            // 关闭线程池
            executor.shutdown();
        }
    }

#### 2.2.2、Future 接口的其他实现类： ####

除了 FutureTask，Java 标准库中还有其他一些类也实现了 Future 接口，例如 ScheduledFuture 接口用于表示可以通过 ScheduledExecutorService 调度执行的任务的结果。

import java.util.concurrent.*;
    
    public class MyScheduledFutureExample {
        
        public static void main(String[] args) {
        
            // 创建调度线程池
            ScheduledExecutorService executor = Executors.newScheduledThreadPool(1);
    
            // 创建一个 Callable 任务
            Callable<Integer> callable = () -> {
        
                System.out.println("Task executed.");
                return 42;
            };
    
            // 调度任务，在 3 秒后执行
            ScheduledFuture<Integer> scheduledFuture = executor.schedule(callable, 3, TimeUnit.SECONDS);
    
            System.out.println("Task scheduled. Waiting for the result...");
    
            try {
        
                // 获取任务的结果，如果任务尚未完成，会阻塞直到任务完成
                int result = scheduledFuture.get();
                System.out.println("Result: " + result);
            } catch (InterruptedException e) {
        
                e.printStackTrace();
            } catch (ExecutionException e) {
        
                Throwable cause = e.getCause(); // 获取实际的异常
                System.err.println("An exception occurred: " + cause.getMessage());
            }
    
            // 关闭线程池
            executor.shutdown();
        }
    }

#### 2.2.3、CompletionStage 接口及其实现类： ####

CompletionStage 是 Java 8 引入的一个用于处理异步操作的接口，它也是 Future 的子接口。CompletionStage 可以表示一个异步计算的阶段，并提供了一系列方法来处理异步操作的结果和完成状态。CompletableFuture 是 CompletionStage 接口的一个实现类，它提供了丰富的方法来处理异步计算，支持流式编程和链式操作。

import java.util.concurrent.CompletableFuture;
    import java.util.concurrent.ExecutionException;
    
    public class MyCompletableFutureExample {
        
        public static void main(String[] args) {
        
            // 创建一个 CompletableFuture
            CompletableFuture<Integer> future = CompletableFuture.supplyAsync(() -> {
        
                System.out.println("Starting async computation...");
                try {
        
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
        
                    e.printStackTrace();
                }
                return 42;
            });
    
            // 处理异步操作的结果和完成状态
            future.thenAccept(result -> {
        
                System.out.println("Result: " + result);
            }).thenRun(() -> {
        
                System.out.println("All processing completed.");
            });
    
            System.out.println("Waiting for the asynchronous computation to complete...");
    
            try {
        
                // 等待异步操作的完成
                System.out.println(future.get());
                System.out.println("try completed");
            } catch (InterruptedException | ExecutionException e) {
        
                e.printStackTrace();
            }
            System.out.println("all completed");
        }
    }

#### 2.2.4、ScheduledFuture： ####

ScheduledFuture 是一个表示可以调度执行的任务的结果的接口，它扩展了 Future 接口。它通常用于与 ScheduledExecutorService 一起使用，用于在指定的时间或延迟后执行任务，并返回任务的结果。

通过使用 Callable 和 Future，可以更好地控制多线程任务的执行和获取结果。这在处理并发编程时非常有用，特别是当需要等待耗时的操作完成并获得结果时。