java并发工具类-Callable、Future 和FutureTask
java并发工具类-Callable、Future 和FutureTask
前言
创建线程的2种方式，一种是直接继承Thread，另外一种就是实现Runnable接口。
这2种方式都有一个缺陷就是：在执行完任务之后无法获取执行结果。
如果需要获取执行结果，就必须通过共享变量或者使用线程通信的方式来达到效果，这样使用起来就比较麻烦。
而自从Java 1.5开始，就提供了Callable和Future，通过它们可以在任务执行完毕之后得到任务执行结果
Callable接口
Callable位于JUC包下，它也是一个接口，在它里面也只声明了一个方法叫做call()：
@FunctionalInterface
public interface Callable<V> {
/**
* Computes a result, or throws an exception if unable to do so.
*
* @return computed result
* @throws Exception if unable to compute a result
*/
V call() throws Exception;
}Callable接口代表一段可以调用并返回结果的代码。
Future接口表示异步任务，是还没有完成的任务给出的未来结果。所以说Callable用于产生结果，Future用于获取结果。
Callable接口使用泛型去定义它的返回类型。
Executors类提供了一些有用的方法在线程池中执行Callable内的任务。由于Callable任务是并行的（并行就是整体看上去是并行的，其实在某个时间点只有一个线程在执行），我们必须等待它返回的结果。
java.util.concurrent.Future对象为我们解决了这个问题。在线程池提交Callable任务后返回了一个Future对象，使用它可以知道Callable任务的状态和得到Callable返回的执行结果。Future提供了get()方法让我们可以等待Callable结束并获取它的执行结果。
那么怎么使用Callable呢？一般情况下是配合ExecutorService来使用的，在ExecutorService接口中声明了若干个submit方法的重载版本：
<T> Future<T> submit(Callable<T> task);
<T> Future<T> submit(Runnable task, T result);
Future<?> submit(Runnable task);第一个方法：submit提交一个实现Callable接口的任务，并且返回封装了异步计算结果的Future。
第二个方法：submit提交一个实现Runnable接口的任务，并且指定了在调用Future的get方法时返回的result对象。
第三个方法：submit提交一个实现Runnable接口的任务，并且返回封装了异步计算结果的Future。
因此我们只要创建好我们的线程对象（实现Callable接口或者Runnable接口），然后通过上面3个方法提交给线程池去执行即可。
Future接口
Future就是对于具体的Runnable或者Callable任务的执行结果进行取消、查询是否完成、获取结果。必要时可以通过get方法获取执行结果，该方法会阻塞直到任务返回结果。
Future接口是用来获取异步计算结果的，说白了就是对具体的Runnable或者Callable对象任务执行的结果进行获取(get())，取消(cancel())，判断是否完成等操作。我们看看Future接口的源码：
public interface Future<V> {
boolean cancel(boolean mayInterruptIfRunning);
boolean isCancelled();
boolean isDone();
V get() throws InterruptedException, ExecutionException;
V get(long timeout, TimeUnit unit)
throws InterruptedException, ExecutionException, TimeoutException;
}在Future接口中声明了5个方法，下面依次解释每个方法的作用
| 方法 | 作用 |
|---|---|
| cance(boolean mayInterruptIfRunning) | 试图取消执行的任务，参数为true时直接中断正在执行的任务，否则直到当前任务执行完成，成功取消后返回true，否则返回false |
| isCancelled() | 方法表示任务是否被取消成功，如果在任务正常完成前被取消成功，则返回 true。 |
| isDone() | 方法表示任务是否已经完成，若任务完成，则返回true； |
| get() | 方法用来获取执行结果，这个方法会产生阻塞，会一直等到任务执行完毕才返回； |
| get(long timeout, TimeUnit unit) | 设定计算结果的返回时间，如果在规定时间内没有返回计算结果则抛出TimeOutException |
Future提供了三种功能
- 判断任务是否完成；
- 能够中断任务；
- 能够获取任务执行结果。
因为Future只是一个接口，所以是无法直接用来创建对象使用的，因此就有了下面的FutureTask。
RunnableFuture接口
RunnableFuture实现了Runnable和Future。因此FutureTask可以传递到线程对象Thread或Excutor(线程池)来执行。
public interface RunnableFuture<V> extends Runnable, Future<V> {
/**
* Sets this Future to the result of its computation
* unless it has been cancelled.
*/
void run();
}FutureTask
我们先来看一下FutureTask的实现：
public class FutureTask<V> implements RunnableFuture<V> FutureTask类实现了RunnableFuture接口，我们看一下RunnableFuture接口的实现
public interface RunnableFuture<V> extends Runnable, Future<V>可以看出RunnableFuture继承了Runnable接口和Future接口，而FutureTask实现了RunnableFuture接口。所以它既可以作为Runnable被线程执行，又可以作为Future得到Callable的返回值。
分析
FutureTask除了实现了Future接口外还实现了Runnable接口，因此FutureTask也可以直接提交给Executor执行。 当然也可以调用线程直接执行（FutureTask.run()）。接下来我们根据FutureTask.run()的执行时机来分析其所处的3种状态：
- 未启动，FutureTask.run()方法还没有被执行之前，FutureTask处于未启动状态，当创建一个FutureTask，而且没有执行FutureTask.run()方法前，这个FutureTask也处于未启动状态。
- 已启动，FutureTask.run()被执行的过程中，FutureTask处于已启动状态。
- 已完成，FutureTask.run()方法执行完正常结束，或者被取消或者抛出异常而结束，FutureTask都处于完成状态。
下面我们再来看看FutureTask的方法执行示意图（方法和Future接口基本是一样的，这里就不过多描述了）
- 当FutureTask处于未启动或已启动状态时，如果此时我们执行FutureTask.get()方法将导致调用线程阻塞；当FutureTask处于已完成状态时，执行FutureTask.get()方法将导致调用线程立即返回结果或者抛出异常。
- 当FutureTask处于未启动状态时，执行FutureTask.cancel()方法将导致此任务永远不会执行。当FutureTask处于已启动状态时，执行cancel(true)方法将以中断执行此任务线程的方式来试图停止任务，如果任务取消成功，cancel(…)返回true；但如果执行cancel(false)方法将不会对正在执行的任务线程产生影响(让线程正常执行到完成)，此时cancel(…)返回false。当任务已经完成，执行cancel(…)方法将返回false。
最后我们给出FutureTask的两种构造函数：
public FutureTask(Callable<V> callable) {
}
public FutureTask(Runnable runnable, V result) {
}事实上，FutureTask是Future接口的一个唯一实现类。
使用创景
通过上面的介绍，我们对Callable，Future，RunnableFuture，FutureTask都有了比较清晰的了解了，那么它们到底有什么用呢？我们前面说过通过这样的方式去创建线程的话，最大的好处就是能够返回结果，加入有这样的场景，我们现在需要计算一个数据，而这个数据的计算比较耗时，而我们后面的程序也要用到这个数据结果，那么这个时Callable岂不是最好的选择？我们可以开设一个线程去执行计算，而主线程继续做其他事，而后面需要使用到这个数据时，我们再使用Future获取不就可以了吗？下面我们就来编写一个这样的实例。
多任务计算
利用FutureTask和ExecutorService，可以用多线程的方式提交计算任务，主线程继续执行其他任务，当主线程需要子线程的计算结果时，在异步获取子线程的执行结果。
例如主线程执行到需要发邮件，发短信的环节，发送完成后才能继续下一次任务，对于发短信发邮件可以使用FutureTask进行多线程进行并行执行，通过get来阻塞返回结果，如果发邮件需要5s，发短信需要8s，传统的方式需要13s，使用FutureTask异步执行只需要8s。
package chapter02.future;
import util.ThreadUtils;
import java.util.concurrent.*;
/**
* 发送消息的FutuerTask
* 将原来发送邮件发送短信的串行任务改为FutuerTask的并行任务提高CPU利用率以及速度
*/
public class FutuerSendMessage {
private static ExecutorService executorService = Executors.newFixedThreadPool(10);
public static void main(String[] args) throws ExecutionException, InterruptedException {
FutuerSendMessage futuerSendMessage = new FutuerSendMessage();
futuerSendMessage.sendmessage();
}
public void sendmessage() throws ExecutionException, InterruptedException {
long currentTime = System.currentTimeMillis();
//发送邮件
Future<Boolean> sendMailFuture = executorService.submit(() -> {
return sendMail();
});
//发送短信
Future<Boolean> sendMsgFuture = executorService.submit(() -> {
return sendMsg();
});
System.out.println("提交任务到线程池，耗时：" + (System.currentTimeMillis() - currentTime));
System.out.println("主线程继续任务...");
//如果发送都成功了
if (sendMailFuture.get() && sendMsgFuture.get()) {
System.out.println("消息发送成功,耗时：" + (System.currentTimeMillis() - currentTime));
}
executorService.shutdown();
}
/**
* 发送邮件
*
* @return
*/
private boolean sendMail() {
//睡眠5s
ThreadUtils.sleep(5, TimeUnit.SECONDS);
System.out.println("发送邮件");
return true;
}
/**
* 发送短信
*
* @return
*/
private boolean sendMsg() {
//睡眠8s
ThreadUtils.sleep(8, TimeUnit.SECONDS);
System.out.println("发送短信");
return true;
}
//-----------------------------原始的代码---------------
public void orginSendmessage() {
long currentTime = System.currentTimeMillis();
boolean sendMail = sendMail();
boolean sendMsg = sendMsg();
if (sendMail && sendMsg) {
System.out.println("消息发送成功,耗时：" + (System.currentTimeMillis() - currentTime));
}
}
}高并发环境下确保任务只执行一次
在很多高并发的环境下，往往我们只需要某些任务只执行一次。这种使用情景FutureTask的特性恰能胜任。举一个例子，假设有一个map的缓存，当key存在时，即直接返回key对应的对象；当key不存在时，则创创建一个对象。对于这样的应用场景，通常采用的方法为使用一个Map对象来存储key和换粗你对象的对应关系，典型的代码如下面所示：
原始的方式如下
private Map<String, String> orginCacheMap = new HashMap<String, String>();
public synchronized String getOrginValue(String key) {
if (cacheMap.containsKey(key)) {
return orginCacheMap.get(key);
} else {
String cacheValue = createCache();
orginCacheMap.putIfAbsent(key, cacheValue);
return cacheValue;
}
}在上面的例子中，我们通过加锁确保高并发环境下的线程安全，也确保了缓存对象只创建一次，然而确牺牲了性能。改用ConcurrentHash的情况下，几乎可以避免加锁的操作，性能大大提高，但是在高并发的情况下有可能出现Connection被创建多次的现象。这时最需要解决的问题就是当key不存在时，创建缓存对象的动作能放在connectionPool之后执行，这正是FutureTask发挥作用的时机，基于ConcurrentHashMap和FutureTask的改造代码如下：
package chapter02.future;
import util.ThreadUtils;
import java.sql.Connection;
import java.util.Collection;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.*;
/**
* FutuerTask 多线程下只进行一次读取
*/
public class FutuerOnceExec {
private Map<String, String> orginCacheMap = new HashMap<String, String>();
private Map<String, Future<String>> cacheMap = new ConcurrentHashMap<String, Future<String>>();
private ExecutorService executorService = Executors.newSingleThreadScheduledExecutor();
public static void main(String[] args) {
FutuerOnceExec futuerOnceExec = new FutuerOnceExec();
long currentTime = System.currentTimeMillis();
String value = futuerOnceExec.getValue("xx");
System.out.println("value:" + value + ",第一次获取耗时：" + (System.currentTimeMillis() - currentTime));
value = futuerOnceExec.getValue("xx");
System.out.println("value:" + value + ",第二次获取耗时：" + (System.currentTimeMillis() - currentTime));
value = futuerOnceExec.getValue("xx");
System.out.println("value:" + value + ",第三次获取耗时：" + (System.currentTimeMillis() - currentTime));
}
/**
* 获取缓存数据
*
* @param key
* @return
*/
public String getValue(String key) {
Future<String> futureValue = cacheMap.get(key);
//如果已经存在Future了直接get 完成的Future是不会阻塞的
if (null != futureValue) {
return getFutureValue(futureValue);
} else {
//使用异步线程常见Future任务
futureValue = executorService.submit(() -> {
ThreadUtils.sleep(1, TimeUnit.SECONDS);
return createCache();
});
executorService.shutdown();
cacheMap.putIfAbsent(key, futureValue);
}
//获取future中的值
return getFutureValue(futureValue);
}
/**
* 获取 Future中的值
* get的时候可能会阻塞
*
* @param futureValue
* @return
*/
private String getFutureValue(Future<String> futureValue) {
try {
return futureValue.get();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
return null;
}
private String createCache() {
//todo 业务代码
return "缓存对象";
}
//--------------------原始的方式
public synchronized String getOrginValue(String key) {
if (cacheMap.containsKey(key)) {
return orginCacheMap.get(key);
} else {
String cacheValue = createCache();
orginCacheMap.putIfAbsent(key, cacheValue);
return cacheValue;
}
}
}总结
实现Runnable接口和实现Callable接口的区别：
- Runnable是自从java1.1就有了，而Callable是1.5之后才加上去的。
- Callable规定的方法是call(),Runnable规定的方法是run()。
- Callable的任务执行后可返回值，而Runnable的任务是不能返回值(是void)。
- call方法可以抛出异常，run方法不可以。
- 运行Callable任务可以拿到一个Future对象，表示异步计算的结果。它提供了检查计算是否完成的方法，以等待计算的完成，并检索计算的结果。通过Future对象可以了解任务执行情况，可取消任务的执行，还可获取执行结果。
- 加入线程池运行，Runnable使用ExecutorService的execute方法，Callable使用submit方法。
Callable、Runnable、Future和FutureTask 的区别
- Callable、Runnable、Future和FutureTask 做为java 线程池运行的重要载体，有必要深入理解。
- Callable 和 Runnable 都是执行的任务的接口，区别在于Callable有返回值，而Runnable无返回值。
- Future 表示异步任务返回结果的接口
- RunnableFuture 继承了Runnable， Future，表示可以带有返回值的run接口
- FutureTask是一个实现类，实现了RunnableFuture接口，既能接受Runnable类型的任务，也可以接受Callable类型的任务，这个类的作用主要是 有一个protected void done()方法用来扩展使用，作为一个回调方法
微信
支付宝
