介绍：是JDK1.5引入的interface，可以方便的使用异步结果的获取；
Future代表的是异步执行的结果，意思是意思是异步线程执行结束之后，返回的结果将保存在Future中；
什么场景适合使用Future呢？
当我们执行长时间运行的任务时，使用Future可以让我们暂时去处理其它事，等待长任务执行完毕返回结果，比如常用的场景是：计算密集型、处理大量数据、远程方法调用等；
Future一般配合Callable接口来使用；
我们将使用ExecutorService中的submit方法来演示Callable和Future的使用
Future submit(Callable task);
Future submit(Runnable task, T result);
Future submit(Runnable task);
代码如下：
public static void main(String[] args) throws Exception {
//线程池中的提供的用法
ExecutorService executorService = Executors.newFixedThreadPool(2);
Future future = executorService.submit(() -> {
String name = Thread.currentThread().getName();
return name;
});
System.out.println(future.get());
//一般单线程的用法
FutureTask futureTask = new FutureTask(() -> {
return Thread.currentThread().getName();
});
new Thread(futureTask).start();
String o = futureTask.get();
System.out.println(o);
}
接下来看看源码，阐述其原理机制：
publicFuture submit(Callable task) { if (task == null) throw new NullPointerException(); //把传入的callable对象封装到FutureTask对象，其间接实现了Runnable 接口，所以可以FutureTask看作是一个普通的Runnable任务； RunnableFuture ftask = newTaskFor(task); //这个就是之前详细说过的ThreadPoolExecutor中的public void execute(Runnable command) 方法了，最会执行FutureTask重写的run()方法； execute(ftask); return ftask; }
FutureTask重写的run()方法：
public void run() {
//还没有被执行的FutureTask的state=NEW
if (state != NEW ||
!UNSAFE.compareAndSwapObject(this, runnerOffset,
null, Thread.currentThread()))
return;
try {
//这个就是拿到我们自己传入的Callable对象
Callable c = callable;
if (c != null && state == NEW) {
V result;
boolean ran;
try {
//调用call()方法,获取返回值
result = c.call();
ran = true;
} catch (Throwable ex) {
result = null;
ran = false;
setException(ex);
}
if (ran)
//把返回结果赋值到FutureTask对象的属性中，并唤醒等待获取结果（调用get()方法）的线程
set(result);
}
} finally {
// runner must be non-null until state is settled to
// prevent concurrent calls to run()
runner = null;
// state must be re-read after nulling runner to prevent
// leaked interrupts
int s = state;
if (s >= INTERRUPTING)
handlePossibleCancellationInterrupt(s);
}
}
protected void set(V v) {
if (UNSAFE.compareAndSwapInt(this, stateOffset, NEW, COMPLETING)) {
//返回结果赋值到FutureTask对象的属性中
outcome = v;
UNSAFE.putOrderedInt(this, stateOffset, NORMAL); // final state
//任务执行完毕，唤醒等待获取结果（调用get()方法）的线程
finishCompletion();
}
}
finishCompletion() 方法
private void finishCompletion() {
// assert state > COMPLETING;
for (WaitNode q; (q = waiters) != null;) {
if (UNSAFE.compareAndSwapObject(this, waitersOffset, q, null)) {
//遍历WaitNode 单项链表，唤醒等待结果的线程
for (;;) {
Thread t = q.thread;
if (t != null) {
q.thread = null;
LockSupport.unpark(t);
}
//等待线程是一个单项链表，目的是防止有多个线程都要获取同一个执行结果；
WaitNode next = q.next;
if (next == null)
break;
q.next = null; // unlink to help gc
q = next;
}
break;
}
}
done();
callable = null; // to reduce footprint
}
在看看FutureTask中的等待获取异步执行结果的get()方法：
public V get() throws InterruptedException, ExecutionException {
int s = state;
//判断当卡状态如果小于COMPLETING状态，也就是当前为NEW状态，则阻塞线程
if (s <= COMPLETING)
s = awaitDone(false, 0L);
//如果为NORMAL状态，则直接返回结果
return report(s);
}
private int awaitDone(boolean timed, long nanos)
throws InterruptedException {
final long deadline = timed ? System.nanoTime() + nanos : 0L;
WaitNode q = null;
boolean queued = false;
for (;;) {
if (Thread.interrupted()) {
removeWaiter(q);
throw new InterruptedException();
}
int s = state;
//如果任务已经完成，则直接返回正常状态，在后面方法中拿到这个状态会直接返回任务结果
if (s > COMPLETING) {
if (q != null)
q.thread = null;
return s;
}
//如果是这个状态，则说明任务已经执行完毕，但还没有把执行结果复制到outcome属性上
else if (s == COMPLETING) // cannot time out yet
Thread.yield();
//说明任务还在执行中，当前没有排队等待执行结果的线程，初始化一个等待线程链表
else if (q == null)
q = new WaitNode();
//加入到线程等待链表中
else if (!queued)
queued = UNSAFE.compareAndSwapObject(this, waitersOffset,
q.next = waiters, q);
//我们可以使用futureTask.get(2,TimeUnit.SECONDS)超时获取结果的方法，如果到时间任务还没执行完毕，则会抛出异常ExecutionException
else if (timed) {
nanos = deadline - System.nanoTime();
if (nanos <= 0L) {
removeWaiter(q);
return state;
}
LockSupport.parkNanos(this, nanos);
}
//阻塞当前线程，等待任务执行完成后被唤醒
else
LockSupport.park(this);
}
}
Fork/Join框架
介绍：ForkJoin是JDK1.7之后提供的多线程并行处理框架；其基本思想是“分而治之”，就是让一个大任务按照给定的阈值分割成多个小任务，然后将多个小任务进行汇总，这样就提高了CUP的使用率；
特性：
1、不是代替ExecutorService，而是对它的补充，某些场景下比ExecutorService性能更好；
2、主要实现了“分而治之”的算法思想；
3、适合计算密集型的任务，因为可以提高CUP使用率；
在JDK8中引入的并行流parallelStream()计算，内部就是采用ForkJoinPool来实现的；
工作窃取算法：指某个线程从其它线程负责的队列里窃取任务来执行；
我们需要执行一个大的任务，我们可以把这个大任务分割成若干个互不依赖的子任务，为了减少线程间获取子任务时的竞争，我们把这些子任务分别放在不同的队列中，并为每个队列创建一个线程来执行里面的任务；但是有的线程会提前把自己负责队列的任务先干完，而其它队列里还有任务等待处理，干完活的线程与其等着，不如帮助其它线程干活，于是它就去其它线程的队列里窃取一个任务来执行；但为了减少窃取任务线程和被窃取任务线程之间在获取任务时的竞争，通常会使用双端队列；
工作窃取法的优点：
充分利用线程进行并行计算，并减少了线程间的竞争；
工作窃取法的缺点：在某些情况下还是会存在竞争的，比如双端队列里只有一个任务时，两个线程同时获取；该算法会消耗更多的系统资源，比如创建多个线程和多个双端队列；
ForkJoinPool的特点：
1、 ForkJoinPool里的每个线程都维护着一个工作队列(WorkQueue)，是一个双端队列，里面存放的对象是任务ForkJoinTask；
2、每个工作线程在运行中产生的新任务（通常是因为调用了fork()方法），会放入工作队列的对尾，并且工作线程处理自己的队列时，使用的是LIFO方式，也就是从队尾获取任务（不过可以设置从对头取任务）；
3、每个工作线程在处理自己的队列时，会尝试窃取其它队列的任务，FIFO的方式；
4、在遇到join时，如果join的任务还未完成，则会先处理其它任务，并等待其子任务完成；
5、在线程自己队列没有任务，也没有可以窃取的任务时，线程会进入休眠状态；
ForkJoin的局限性：
1、 在遇到join时，如果join的任务还未完成，则会先处理其它任务，并等待其子任务完成，如果一个子任务进行了阻塞，那么这个工作线程就不能执行其它任务；
2、所拆分的任务不应该去执行IO操作，比如读写数据文件；
3、任务不能抛出异常，否则会中断所有任务；
ForkJoinTask：
封装任务的类，里面定义里拆分任务、获取任务、合并子任务结果的时机和规则；
里面有两个重要的方法：
fork()：拆分任务；
join()：等待子任务的结果
通常情况下我们不需要继承抽象类ForkJoinTask，只需要继承它的子类就可以了，JDK提供了很多子类，但是常用的只有三个：
RecursiveTask：有返回结果的任务。（可以将大任务分割成多个小任务，并将这些子任务的结果合并到一个结果集）；
RecursiveAction：用于没有返回结果的任务，比如写数据到磁盘（不建议）；
CountedCompleter：在任务完成之后会触发执行自定义的钩子函数；
下面是ForkJoin使用的Demo:
需求：计算一个数据段中所有整数相加的总和：
public class ForkJoinTaskDemo extends RecursiveTask{ private static final int threshold = 3; private int start; private int end; public ForkJoinTaskDemo(int start, int end) { this.start = start; this.end = end; } @Override protected Long compute() { long sum = 0; if (end - start <= threshold) { for (int i = start; i <= end; i++) { sum += i; } } else { int middle = (start + end) / 2; ForkJoinTaskDemo leftTask = new ForkJoinTaskDemo(start, middle); ForkJoinTaskDemo rightTask = new ForkJoinTaskDemo(middle + 1, end); //添加子任务到队列中 leftTask.fork(); rightTask.fork(); //等待子任务结果 long leftResult = leftTask.join(); long rightResult = rightTask.join(); //合并子任务结果 sum = leftResult + rightResult; } return sum; } public static void main(String[] args) throws ExecutionException, InterruptedException { int start = 1; int end = 9999999; //默认线程数为CPU核数 ForkJoinPool forkJoinPool = new ForkJoinPool(); ForkJoinTaskDemo joinTask = new ForkJoinTaskDemo(start, end); ForkJoinTask submit = forkJoinPool.submit(joinTask); Long result = submit.get(); System.out.println("执行结果为:" + result); forkJoinPool.shutdown(); } }
