Java并发----Callable、Future和FutureTask

逃离我推掉我的手 2022-07-14 23:37 234阅读 0赞

## Callable、Future和FutureTask    ##

创建线程一般使用Thread或者实现Runnable接口，这种缺陷是执行完毕之后是无法获取到结果的。（run()返回是void）。

如果想获取到结果需要使用共享变量或者线程之间的通信获取，如此做不仅繁琐对编程能力有要求比较高。

从JAVA1.5提供了Callable、Future来使线程执行完毕之后返回结果。

下面是我总结的使用Callable、Future和FutureTask的使用方法：

1.Callable位于java.util.concurrent包，里面只声明了一个方法call();(类似run()，但是它是有返回结果类型的--V)；

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;
 }

一般配合ExecutorService类来使用，这个类中有submit方法：

<T> Future<T> submit(Callable<T> task);
 <T> Future<T> submit(Runnable task, T result);
 Future<?> submit(Runnable task);

一般情况下我们只使用第一个和第三个，第二个很少使用。

它们返回的都是Future，我们来看看这个是怎么使用的。

2.Future

Future就是对Callable和Runnable任务的执行结果进行取消、查询是否完成、获取结果等操作。必要时可通过get()获取结果，任务阻塞直到任务返回结果。

Future类位于java.util.concurrent包下：

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;
 }

cancel方法用来取消任务，如果任务取消则返回成功true，否则返回fasle。-----mayInterruptIfRunning表示是否允许取消正在执行但没有执行完毕的任务，如果设置true，则表示允许取消。

isCancelled方法表示任务是否被取消成功。

isDone方法表示任务是否已经完成。

get方法表示用来获取执行结果，这个方法会产生阻塞，会一直等到任务执行完毕才返回。

get（long timeout,TimeUnit unit）用来获取执行结果，如果在指定时间内未获取到则直接返回null。

此接口定义了许多对线程执行结果的一些验证和返回。具体实现类FutureTask才是核心。

3.FutureTask

FutureTask是Future接口的唯一实现类。

它实现了RunnableFuture接口，而这接口又继承了Runnable和Future接口。所以它可作为Runnable被线程执行，又可以作为Callable的返回值。

\-------------------------------------------  使用案例  -------------------------------------------------

使用Callable和Future获取执行结果：

List<Map<String, Object>> list = new ArrayList<Map<String,Object>>();
 		Map<String, Object> params = new HashMap<String, Object>();
 		params.put("taskName", "结果1");
 		list.add(params);
 		params = new HashMap<String, Object>();
 		params.put("taskName", "结果2");
 		list.add(params);
 		params = new HashMap<String, Object>();
 		params.put("taskName", "结果3");
 		list.add(params);
 		params = new HashMap<String, Object>();
 		params.put("taskName", "结果4");
 		list.add(params);
 		params = new HashMap<String, Object>();
 		params.put("taskName", "结果5");
 		list.add(params);
 		params = new HashMap<String, Object>();
 		params.put("taskName", "结果6");
 		list.add(params);
 		params = new HashMap<String, Object>();
 		params.put("taskName", "结果7");
 		list.add(params);
 		params = new HashMap<String, Object>();
 		params.put("taskName", "结果8");
 		list.add(params);
 		params = new HashMap<String, Object>();
 		params.put("taskName", "结果9");
 		list.add(params);
 		params = new HashMap<String, Object>();
 		params.put("taskName", "结果10");
 		list.add(params);
 		params = new HashMap<String, Object>();
 		params.put("taskName", "结果11");
 		list.add(params);
 		params = new HashMap<String, Object>();
 		params.put("taskName", "结果12");
 		list.add(params);
 		params = new HashMap<String, Object>();
 		params.put("taskName", "结果13");
 		list.add(params);
 		params = new HashMap<String, Object>();
 		params.put("taskName", "结果14");
 		list.add(params);
 		params = new HashMap<String, Object>();
 		params.put("taskName", "结果15");
 		list.add(params);
 		params = new HashMap<String, Object>();
 		params.put("taskName", "结果16");
 		list.add(params);
 		params = new HashMap<String, Object>();
 		params.put("taskName", "结果17");
 		list.add(params);
 		params = new HashMap<String, Object>();
 		params.put("taskName", "结果18");
 		list.add(params);
 		params = new HashMap<String, Object>();
 		params.put("taskName", "结果19");
 		list.add(params);
 		params = new HashMap<String, Object>();
 		params.put("taskName", "结果20");
 		list.add(params);
 		Task<String> task = new SendTask();
 		long start = System.currentTimeMillis();
 		List<Future<String>> results = TaskHandler.dealTask(list, task,10000);
 		long end = System.currentTimeMillis();
 		System.out.println(end-start);
 		int i=0;
 		for (Future<String> f : results) {
 			i++;
 			System.out.println("返回结果："+i);
 			try {
 				System.out.println(f.get());
 			} catch (Exception e) {
 				e.printStackTrace();
 			}
 		}

public class SendTask implements Task<String>{
 	
 	private Map<String, Object> params;
 	
 	@Override
 	public void taskParams(Map<String, Object> params) {
 		this.params = params;
 	}
 
 
 	@Override
 	public String call() throws Exception {
 		System.out.println("线程等待2s");
 		Thread.sleep(2000);
 		System.out.println(params.get("taskName"));
 		return "返回成功！";
 	}
 
 
 }

public static <T>List<Future<T>> dealTask(List<Map<String, Object>> params,Task<T> task,long timeOut){
 		List<Future<T>> list = new ArrayList<Future<T>>();
 		//获取缓存线程池
 		ExecutorService service = ThreadLocalHelper.getCachedPool();
 		// 执行多任务,循环处理参数集
 		Future<T> future = null;
 		//初始化返回结果集
 		for (Map<String, Object> param : params) {
 			try {
 				// 给任务设置任务参数
 				task.taskParams(param);
 				//执行任务
 				future = service.submit(task);
 				list.add(future);
 			} catch (Exception e) {
 				e.printStackTrace();
 			} 
 		}
 		//执行完毕后关闭线程池
 		service.shutdown();
 		//获取开始时间（若超过20S，则强制退出返回）
 		long start = System.currentTimeMillis();
 		long flag = 0;
 		while(true){
 			flag = System.currentTimeMillis()-start;
 			if(service.isTerminated()||flag>=timeOut){
 				System.out.println("线程执行完毕!");
 				break;
 			}
 			try {
 				Thread.sleep(10);
 			} catch (Exception e) {
 				e.printStackTrace();
 			}
 		}
 		return list;
 	}