java Callable & Future & FutureTask

痛定思痛。 2021-10-19 04:56 356阅读 0赞

### 文章目录 ###

*  前言
 *  一、Callable与Runnable
 *  二、Future接口
 *  三、FutureTask实现类
 *  四、使用示例
 *  感谢

# 前言 #

实现Runnable接口的线程类与一个缺陷，就是在任务执行完之后无法取得任务的返回值。 如果需要获取执行结果，就必须通过共享变量或者使用线程通信的方式来达到效果，这样使用起来就比较麻烦 。所以，从JDK 1.5开始，java提供了Callable接口，该接口和Runnable接口相类似，提供了一个call()方法可以作为线程的执行体，但是call()方法要比run()方法更为强大：\# call()方法可以有返回值；\# call()方法可以声明抛出异常。那么使用callable接口是如何获取返回值的呢？

# 一、Callable与Runnable #

既然Callable接口可以看做是Runnable的“增强版”，那我们先看看Runnable接口的实现，追根溯源也是搬砖的普遍素质嘛~

public interface Runnable { 
        public abstract void run();
    }

Runnable接口中只包含一个抽象方法run()返回值为void， 所以在执行完任务之后无法返回任何结果。接下来我们可以看看Callable接口

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

这是一个泛型接口，call()函数返回的类型就是传递进来的V类型。 那么该如何使用Callable接口呢?Callable接口不是Runnable接口的子接口，所以Callable对象不能直接作为Thread的target去运行；而且call方法还有返回值–call()方法并不是直接调用。

JDK提供了Future接口来代表call()方法里的返回值，并且为Future接口提供了一个实现类FutureTask，该类实现了Future接口，并实现了Runnable接口，其实该类也是Future接口的唯一实现类，所以FutureTask对象可以作为Thread的target。Callable一般配合ExecutorService来使用的，在ExecutorService接口中声明了若干个submit方法的重载版本：

/** * Submits a value-returning task for execution and returns a * Future representing the pending results of the task. The * Future's <tt>get</tt> method will return the task's result upon * successful completion. */
 <T> Future<T> submit(Callable<T> task);
 
 /** * Submits a Runnable task for execution and returns a Future * representing that task. The Future's <tt>get</tt> method will * return the given result upon successful completion. */
 <T> Future<T> submit(Runnable task, T result);
 
 /** * Submits a Runnable task for execution and returns a Future * representing that task. The Future's <tt>get</tt> method will * return <tt>null</tt> upon successful completion. */
 Future<?> submit(Runnable task);

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

**这里提一下ExecutorService的submit与execute方法的区别：**

ExecutorService的submit与execute方法都能执行任务，但在使用过程，发现其对待run方法抛出的异常处理方式不一样。两者执行任务最后都会通过Executor的execute方法来执行，但对于submit，会将runnable物件包装成FutureTask,其run方法会捕捉被包装的Runnable Object的run方法抛出的Throwable异常，待submit方法所返回的的Future Object调用get方法时，将执行任务时捕获的Throwable Object包装成java.util.concurrent.ExecutionException来抛出。  
而对于execute方法，则会直接抛出异常，该异常不能被捕获，想要在出现异常时做些处理，可以实现Thread.UncaughtExceptionHandler接口。

当将一个Callable的对象传递给ExecutorService的submit方法，则该call方法自动在一个线程上执行，并且会返回执行结果Future对象。同样，将Runnable的对象传递给ExecutorService的submit方法，则该run方法自动在一个线程上执行，并且会返回执行结果Future对象，但是在该Future对象上调用get方法，将返回null。

# 二、Future接口 #

Future就是对于具体的Runnable或者Callable任务的执行结果进行取消、查询是否完成、获取结果。必要时可以通过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;
 }

在Future接口中声明了5个方法，下面依次解释每个方法的作用：

*  cancel方法用来取消任务，如果取消任务成功则返回true，如果取消任务失败则返回false。参数 mayInterruptIfRunning表示是否允许取消正在执行却没有执行完毕的任务，如果设置true，则表示可以取消正在执行过程中的任务。如果任务已经完成，则无论mayInterruptIfRunning为true还是false，此方法肯定返回false，即如果取消已经完成的任务会返 回false；如果任务正在执行，若mayInterruptIfRunning设置为true，则返回true，若 mayInterruptIfRunning设置为false，则返回false；如果任务还没有执行，则无论 mayInterruptIfRunning为true还是false，肯定返回true。
 *  isCancelled方法表示任务是否被取消成功，如果在任务正常完成前被取消成功，则返回 true。
 *  isDone方法表示任务是否已经完成，若任务完成，则返回true；
 *  get()方法用来获取执行结果，这个方法会产生阻塞，会一直等到任务执行完毕才返回；
 *  get(long timeout, TimeUnit unit)用来获取执行结果，如果在指定时间内，还没获取到结果，就直接返回null。

也就是说Future提供了三种功能：

1）判断任务是否完成；

2）能够中断任务；

3）能够获取任务执行结果。

因为Future只是一个接口，所以是无法直接用来创建对象使用的，因此就有了FutureTask。

# 三、FutureTask实现类 #

我们先来看一下FutureTask的实现：

public class FutureTask<V> implements RunnableFuture<V>

FutureTask类实现了RunnableFuture接口，我们再来看一下RunnableFuture接口的实现：

public interface RunnableFuture<V> extends Runnable, Future<V> { 
 void run();
 }

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

FutureTask提供了2个构造器：

public FutureTask(Callable<V> callable) { 
 }
 public FutureTask(Runnable runnable, V result) { 
 }

如何有童鞋对FutureTask具体内容感兴趣，可以看一下这篇文章：[FutureTask深入解析][FutureTask]

# 四、使用示例 #

package com.blog.www.thread;
 
 import com.blog.www.util.ListUtils;
 import com.google.common.collect.Lists;
 import com.google.common.util.concurrent.ThreadFactoryBuilder;
 
 import java.util.List;
 import java.util.concurrent.*;
 
 /** * 第一个线程 * * * @author ：leigq * @date ：2019/8/13 16:12 */
 public class FirstThread implements Callable<String> { 
 
 	private List<String> list;
 
 	public FirstThread(List<String> list) { 
 		this.list = list;
 	}
 
 	@Override
 	public String call() { 
 		System.out.println(list);
 		return String.format("My name is %s", Thread.currentThread().getName());
 	}
 }
 
 class Test{ 
 	public static void main(String[] args) { 
 		/* * 多线程使用思路： * 1、查询数据，每次查询 dataCount 条 * 2、开 threadCount 个线程处理数据，每个线程处理 dataCount / threadCount 条 * */
 		int dataCount = 100, threadCount = 20, maxThreadCount = 30;
 		// 使用 ThreadPoolExecutor 创建线程池，阿里 java 开发手册推荐，详见：https://blog.csdn.net/fly910905/article/details/81584675
 		ThreadFactory threadFactory = new ThreadFactoryBuilder().setNameFormat("test-pool-%d").build();
 		ExecutorService pool = new ThreadPoolExecutor(threadCount, maxThreadCount, 0L, TimeUnit.MILLISECONDS, new LinkedBlockingQueue<>(1024), threadFactory, new ThreadPoolExecutor.AbortPolicy());
 
 		// 模拟 dataCount 条数据
 		List<String> strings = Lists.newArrayList();
 		for (int i = 0; i < dataCount; i++) { 
 			strings.add(String.format("data %d", i));
 		}
 		// 平分数据
 		List<List<String>> lists = ListUtils.avgAssign(strings, threadCount);
 		// 保存执行结果
 		List<FutureTask<String>> futureTasks = Lists.newArrayList();
 		lists.forEach(list -> { 
 			FirstThread firstThread = new FirstThread(list);
 			FutureTask<String> futureTask = new FutureTask<>(firstThread);
 			pool.submit(futureTask);
 			futureTasks.add(futureTask);
 		});
 
 		futureTasks.forEach(futureTask -> { 
 			try { 
 				System.out.println(futureTask.get());
 			} catch (InterruptedException | ExecutionException e) { 
 				e.printStackTrace();
 			}
 		});
 		pool.shutdown();
 	}
 }

执行结果：

[data 0, data 1, data 2, data 3, data 4]
    [data 10, data 11, data 12, data 13, data 14]
    [data 5, data 6, data 7, data 8, data 9]
    [data 15, data 16, data 17, data 18, data 19]
    [data 20, data 21, data 22, data 23, data 24]
    [data 25, data 26, data 27, data 28, data 29]
    [data 30, data 31, data 32, data 33, data 34]
    [data 35, data 36, data 37, data 38, data 39]
    [data 40, data 41, data 42, data 43, data 44]
    [data 45, data 46, data 47, data 48, data 49]
    [data 50, data 51, data 52, data 53, data 54]
    [data 55, data 56, data 57, data 58, data 59]
    [data 60, data 61, data 62, data 63, data 64]
    [data 65, data 66, data 67, data 68, data 69]
    [data 70, data 71, data 72, data 73, data 74]
    [data 75, data 76, data 77, data 78, data 79]
    [data 80, data 81, data 82, data 83, data 84]
    [data 85, data 86, data 87, data 88, data 89]
    [data 90, data 91, data 92, data 93, data 94]
    [data 95, data 96, data 97, data 98, data 99]
    My name is test-pool-0
    My name is test-pool-1
    My name is test-pool-2
    My name is test-pool-3
    My name is test-pool-4
    My name is test-pool-5
    My name is test-pool-6
    My name is test-pool-7
    My name is test-pool-8
    My name is test-pool-9
    My name is test-pool-10
    My name is test-pool-11
    My name is test-pool-12
    My name is test-pool-13
    My name is test-pool-14
    My name is test-pool-15
    My name is test-pool-16
    My name is test-pool-17
    My name is test-pool-18
    My name is test-pool-19
    
    Process finished with exit code 0

上面用到了一个`ListUtils.avgAssign(List<T> list, Integer n)`方法，如下：

package com.blog.www.util;
 
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 
 /** * 集合工具类 * * * @author ：leigq * @date ：2019/8/13 14:16 */
 public class ListUtils { 
 
 	/** * 按指定大小，分隔集合，将集合按规定个数分为 n 个部分 * * create by: leigq * * create time: 2019/8/13 14:16 * * @param list : 待拆分的集合 * @param elementCount : 元素个数 * @return 拆分后的集合 */
 	public static <T> List<List<T>> split(List<T> list, Integer elementCount) { 
 		if (list == null || list.isEmpty() || elementCount < 1) { 
 			return Collections.emptyList();
 		}
 		List<List<T>> result = new ArrayList<>();
 		int size = list.size();
 		int count = (size + elementCount - 1) / elementCount;
 		for (int i = 0; i < count; i++) { 
 			List<T> subList = list.subList(i * elementCount, (Math.min((i + 1) * elementCount, size)));
 			result.add(subList);
 		}
 		return result;
 	}
 
 	/** * 将一个list均分成 n 个list * * create by: leigq * * create time: 2019/8/13 14:11 * * @param list : 待拆分的集合 * @param n : 拆分数量 * @return 拆分后的集合， if list.size > n, 平分或者前几个集合有多个元素; if list.size < n, 后几集合为空 */
 	public static <T> List<List<T>> avgAssign(List<T> list, Integer n) { 
 		List<List<T>> result = new ArrayList<>();
 		// 先计算出余数
 		int remainder = list.size() % n;
 		// 然后是商
 		int number = list.size() / n;
 		//偏移量
 		int offset = 0;
 		for (int i = 0; i < n; i++) { 
 			List<T> value;
 			if (remainder > 0) { 
 				value = list.subList(i * number + offset, (i + 1) * number + offset + 1);
 				remainder--;
 				offset++;
 			} else { 
 				value = list.subList(i * number + offset, (i + 1) * number + offset);
 			}
 			result.add(value);
 		}
 		return result;
 	}
 }

还有一点需要注意，就是在线程类中注入`Spring`的`bean`：详见：https://www.cnblogs.com/xuyuanjia/p/6097875.html

# 感谢 #

*  [java Callable & Future & FutureTask][java Callable _ Future _ FutureTask]
 *  [ThreadPoolExecutor：线程池不允许使用Executors创建][ThreadPoolExecutor_Executors]

[FutureTask]: https://blog.csdn.net/liulipuo/article/details/39029643
[java Callable _ Future _ FutureTask]: https://my.oschina.net/cain1507/blog/747258
[ThreadPoolExecutor_Executors]: https://blog.csdn.net/fly910905/article/details/81584675