高级并发编程学习-callable与Future的使用

应用场景：FutureTask可用于异步获取执行结果或取消执行任务的场景。通过传入Runnable或者Callable的任务给FutureTask，直接调用其run方法或者放入线程池执行，之后可以在外部通过FutureTask的get方法异步获取执行结果，因此，FutureTask非常适合用于耗时的计算，主线程可以在完成自己的任务后，再去获取结果。另外，FutureTask还可以确保即使调用了多次run方法，它都只会执行一次Runnable或者Callable任务，或者通过cancel取消FutureTask的执行等。

注意：Callable接口不能替代Runnable，原因是Callable必须要和线程池Executors结合使用。Future取得的结果类型和callable返回的结果类型必须一致，通过泛型来实现。Callable要采用ExceutorService的submit方法提交，返回的future对象可以取消任务。FutureTask是接口Future的唯一的实现类。

使用步骤：

       
(1)  任务需要实现callable接口，注意这里的泛型：

@Override
public Integer call() throws Exception {
Integer result=0;
return result;
}

(2)生成任务并提交，存在两种方式。
         第一种：
 

Future<String> future=threadPool.submit(new Callable<String>(){
@Override
public String call() throws Exception {
Thread.sleep(2000);
return "hello";
}
});

使用例子：

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

public class CallableAndFuture {
public static void main(String[] args) {
ExecutorService threadPool=Executors.newSingleThreadExecutor();
//future的泛型类型必须要与callable的泛型类型相一致
Future<String> future=threadPool.submit(new Callable<String>(){

@Override
public String call() throws Exception {
Thread.sleep(2000);
return "hello";
}

});
System.out.println("等待结果....");
try {
//future.get方法会一直等待，也会阻塞线程
System.out.println("拿到结果：future.get方法的结果是"+future.get());
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
} catch (ExecutionException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
System.out.println("aaaaaaaaaaaa");
threadPool.shutdown();//必须显示关闭
}

}

第二种：

FutureTask<Integer> ft = new FutureTask<Integer>(new ComputeTask(i, ""+i));
exec.submit(ft);

注意这里必须为FutureTask而非Future，因为ExecutorService.submit（）方法声明为：
Future<?>submit(Runnabletask)或
Future<T> submit(Callable<T>task);

这就要求提交的任务需要实现了Runnable接口或Callable接口，因此这里需要使用FutureTask. 



示例代码：

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.FutureTask;

public class FutureTaskForMultiCompute {

public static void main(String[] args) {

FutureTaskForMultiCompute inst=new FutureTaskForMultiCompute();
// 创建任务集合
List<Future<Integer>> taskList = new ArrayList<Future<Integer>>();
// 创建线程池
ExecutorService exec = Executors.newFixedThreadPool(5);
for (int i = 0; i < 10; i++) {
//传入Callable对象创建FutureTask注意这里的ft实现了callable接口
FutureTask<Integer> ft = new FutureTask<Integer>(new ComputeTask(i, ""+i));
/*第二种提交方式为
Future<Integer>ft=exec.submit(new ComputeTask(i,""+i));*/
taskList.add(ft);
//提交给线程池执行任务，也可以通过exec.invokeAll(taskList)一次性提交所有任务;
exec.submit(ft);
}

System.out.println("所有计算任务提交完毕, 主线程接着干其他事情！");

// 开始统计各计算线程计算结果
Integer totalResult = 0;
for (Future<Integer> ft : taskList) {
try {
//FutureTask的get方法会自动阻塞,直到获取计算结果为止
totalResult = totalResult + ft.get();
} catch (InterruptedException e) {
e.printStackTrace();
} catch (ExecutionException e) {
e.printStackTrace();
}
}

// 关闭线程池
exec.shutdown();
System.out.println("多任务计算后的总结果是:" + totalResult);

}

}

import java.util.concurrent.Callable;

public class ComputeTask implements Callable<Integer> {

private Integer result = 0;
private String taskName = "";

public ComputeTask(Integer iniResult, String taskName){
result = iniResult;
this.taskName = taskName;
System.out.println("生成子线程计算任务: "+taskName);
}

public String getTaskName(){
return this.taskName;
}

@Override
public Integer call() throws Exception {
// TODO Auto-generated method stub

for (int i = 0; i < 100; i++) {
result =+ i;
}
// 休眠5秒钟，观察主线程行为，预期的结果是主线程会继续执行，到要取得FutureTask的结果是等待直至完成。
Thread.sleep(5000);
System.out.println("子线程计算任务: "+taskName+" 执行完成!");
return result;
}
}