Java多线程 - 利用Callable或CompletableFuture实现多线程异步任务执行

1. Callable接口源码

@FunctionalInterface
public interface Callable<V> {// 这个call()方法有返回值，且声明了受检异常，可以直接抛出Exception异常V call() throws Exception;
}

2. 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接口中的方法：

get()：获取异步任务执行的结果。注意，这个方法的调用是阻塞性的。如果异步任务没有执行完成，异步结果获取线程（调用线程）会一直被阻塞，一直阻塞到异步任务执行完成，其异步结果返回给调用线程。

get(Long timeout,TimeUnit unit)：设置时限，（调用线程）阻塞性地获取异步任务执行的结果。该方法的调用也是阻塞性的，但是结果获取线程（调用线程）会有一个阻塞时长限制，不会无限制地阻塞和等待，如果其阻塞时间超过设定的timeout时间，该方法将抛出异常，调用线程可捕获此异常。

boolean isDone()：获取异步任务的执行状态。如果任务执行结束，就返回true。

boolean isCancelled()：获取异步任务的取消状态。如果任务完成前被取消，就返回true。

boolean cancel(boolean mayInterruptRunning)：取消异步任务的执行。

3. RunnableFuture接口和FutureTask实现类

如何使用Callable接口创建线程呢？

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

RunnableFuture只是一个接口，无法直接创建对象，如果需要创建对象，就需用到它的实现类——FutureTask。

public class FutureTask<V> implements RunnableFuture<V> {private Callable<V> callable;// 构造方法public FutureTask(Callable<V> callable) {if (callable == null) throw new NullPointerException();// callable实例属性需要在FutureTask实例构造时进行初始化this.callable = callable;this.state = NEW;       }
}

RunnableFuture接口实现了2个目标：

(1) RunnableFuture通过继承Runnable接口，从而保证了其实例可以作为Thread线程实例的target目标；
(2) RunnableFuture通过继承Future接口，从而保证了可以获取未来的异步执行结果。

首先，通过实现Runnable接口的方式创建一个异步执行任务：

public class CallableTaskDemo implements Callable {// call()方法有返回值，并且可以抛出Exception异常@Overridepublic String call() throws Exception {System.out.println("实现Callable接口来编写异步执行任务");Thread.sleep(1000);return "返回线程执行结果";}
}

方式1：通过Thread类创建线程执行异步任务

public class CallableDemo {public static void main(String[] args) throws ExecutionException, InterruptedException {// 创建异步执任务实例Callable callable = new CallableTaskDemo();// 初始化callable实例属性FutureTask futureTask = new FutureTask(callable);// 创建线程执行异步任务Thread thread = new Thread(futureTask);thread.start();System.out.println("获取异步执行任务结果："+futureTask.get());// 获取异步执行任务结果：返回线程执行结果}
}

方式2：通过线程池创建线程，并提交异步执行任务：

public class CallableDemo2 {// 通过线程池创建3个线程private static ExecutorService executorService = Executors.newFixedThreadPool(3);public static void main(String[] args) throws ExecutionException, InterruptedException {// 通过线程池的submit()方法提交异步执行任务，有返回结果Future submit = executorService.submit(new CallableTaskDemo());// 获取返回结果System.out.println(submit.get());}
}

对于Calleble来说，Future和FutureTask均可以用来获取任务执行结果，不过Future是个接口，FutureTask是Future的具体实现，而且FutureTask还间接实现了Runnable接口，也就是说FutureTask可以作为Runnable任务提交给线程池。

4. 利用线程池和Callable接口实现异步执行任务

1、定义3个线程执行任务：

public class FirstCallableTask implements Callable<String> {// call()方法有返回值，并且可以抛出Exception异常@Overridepublic String call() throws Exception {System.out.println("实现 First Callable接口来编写异步执行任务");Thread.sleep(1000);return "返回 First Callable 接口线程执行结果";}
}public class SecondCallableTask implements Callable<String> {@Overridepublic String call() throws Exception {System.out.println("实现 Second Callable接口来编写异步执行任务");Thread.sleep(1000);return "返回 Second Callable 接口线程执行结果";}
}public class ThirdCallableTask implements Callable<String> {@Overridepublic String call() throws Exception {System.out.println("实现 Third Callable接口来编写异步执行任务");Thread.sleep(1000);return "返回 Third Callable 接口线程执行结果";}
}

2、定义线程池提交线程任务：

@Service
@Slf4j
public class BeanLoadService implements ApplicationContextAware, ApplicationListener<ContextStoppedEvent> {private ApplicationContext applicationContext;// 定义一个线程池private static final ThreadPoolExecutor THREAD_POOL_EXECUTOR = new ThreadPoolExecutor(5,7,4,TimeUnit.SECONDS,new LinkedBlockingQueue<>(),new ThreadFactoryBuilder().setNamePrefix(BeanLoadService.class.getSimpleName() + "-pool-%d").setDaemon(true).build(),new ThreadPoolExecutor.DiscardOldestPolicy());public Map<String,Integer> richInfo() {Callable<String> firstCallable = new FirstCallableTask();Callable<String> secondCallable = new SecondCallableTask();Callable<String> thirdCallable = new ThirdCallableTask();List<Callable<String>> callableList = new ArrayList<>();callableList.add(firstCallable);callableList.add(secondCallable);callableList.add(thirdCallable);try {List<Future<String>> futures = THREAD_POOL_EXECUTOR.invokeAll(callableList, 1, TimeUnit.MINUTES);for (Future<String> future : futures) {try {System.out.println(future.get());} catch (Exception e) {log.warn("incident delay data,future get warn", e);}}} catch (Exception e) {log.warn("incident delay data warn ", e);}}@Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {this.applicationContext = applicationContext;}@Overridepublic void onApplicationEvent(ContextStoppedEvent contextStoppedEvent) {try {THREAD_POOL_EXECUTOR.shutdown();} catch (Exception e) {log.error("停止线程池失败", e);}}
}

3、测试：

@SpringBootTest
@RunWith(SpringRunner.class)
public class BeanLoadServiceTest {@Autowiredprivate BeanLoadService beanLoadService;@Testpublic void test(){beanLoadService.richInfo();}
}

实现 First Callable接口来编写异步执行任务
实现 Second Callable接口来编写异步执行任务
实现 Third Callable接口来编写异步执行任务
返回 First Callable 接口线程执行结果
返回 Second Callable 接口线程执行结果
返回 Third Callable 接口线程执行结果

4、定义线程池提交线程任务（内部类）：

@Service
@Slf4j
public class BeanLoadService implements ApplicationContextAware, ApplicationListener<ContextStoppedEvent> {private ApplicationContext applicationContext;// 定义一个线程池private static final ThreadPoolExecutor THREAD_POOL_EXECUTOR = new ThreadPoolExecutor(5,7,4,TimeUnit.SECONDS,new LinkedBlockingQueue<>(),new ThreadFactoryBuilder().setNamePrefix(BeanLoadService.class.getSimpleName() + "-pool-%d").setDaemon(true).build(),new ThreadPoolExecutor.DiscardOldestPolicy());public Map<String,Integer> richInfo() {Callable<String> firstCallable = getFirstCallable();Callable<String> secondCallable = getSecondCallable();Callable<String> thirdCallable = getThirdCallable();List<Callable<String>> callableList = new ArrayList<>();callableList.add(firstCallable);callableList.add(secondCallable);callableList.add(thirdCallable);try {List<Future<String>> futures = THREAD_POOL_EXECUTOR.invokeAll(callableList, 1, TimeUnit.MINUTES);for (Future<String> future : futures) {try {System.out.println(future.get());} catch (Exception e) {log.warn("incident delay data,future get warn", e);}}} catch (Exception e) {log.warn("incident delay data warn ", e);}}private Callable<String> getThirdCallable() {Callable<String> callable = new Callable<String>() {@Overridepublic String call() throws Exception {System.out.println("实现 Third Callable接口来编写异步执行任务");Thread.sleep(1000);return "返回 Third Callable 接口线程执行结果";}};}private Callable<String> getSecondCallable() {Callable<String> callable = new Callable<String>() {@Overridepublic String call() throws Exception {System.out.println("实现 Second Callable接口来编写异步执行任务");Thread.sleep(1000);return "返回 Second Callable 接口线程执行结果";};};return callable;}private Callable<String> getFirstCallable() {Callable<String> callable = new Callable<String>() {@Overridepublic String call() throws Exception {System.out.println("实现 First Callable接口来编写异步执行任务");Thread.sleep(1000);return "返回 First Callable 接口线程执行结果";}};return callable;}@Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {this.applicationContext = applicationContext;}@Overridepublic void onApplicationEvent(ContextStoppedEvent contextStoppedEvent) {try {THREAD_POOL_EXECUTOR.shutdown();} catch (Exception e) {log.error("停止线程池失败", e);}}
}

5、定义线程池提交任务：（Java 8）

@Service
@Slf4j
public class BeanLoadService implements ApplicationContextAware, ApplicationListener<ContextStoppedEvent> {private ApplicationContext applicationContext;// 定义一个线程池private static final ThreadPoolExecutor THREAD_POOL_EXECUTOR = new ThreadPoolExecutor(5,7,4,TimeUnit.SECONDS,new LinkedBlockingQueue<>(),new ThreadFactoryBuilder().setNamePrefix(BeanLoadService.class.getSimpleName() + "-pool-%d").setDaemon(true).build(),new ThreadPoolExecutor.DiscardOldestPolicy());public void richInfo() {Callable<String> firstCallable = getFirstCallable();Callable<String> secondCallable = getSecondCallable();Callable<String> thirdCallable = getThirdCallable();List<Callable<String>> callableList = new ArrayList<>();callableList.add(firstCallable);callableList.add(secondCallable);callableList.add(thirdCallable);try {List<Future<String>> futures = THREAD_POOL_EXECUTOR.invokeAll(callableList, 1, TimeUnit.MINUTES);for (Future<String> future : futures) {try {System.out.println(future.get());} catch (Exception e) {log.warn("incident delay data,future get warn", e);}}} catch (Exception e) {log.warn("incident delay data warn ", e);}}private Callable<String> getThirdCallable() {Callable<String> callable = ()->{System.out.println("实现 Third Callable接口来编写异步执行任务");Thread.sleep(1000);return "返回 Third Callable 接口线程执行结果";};return callable;}private Callable<String> getSecondCallable() {Callable<String> callable = ()-> {System.out.println("实现 Second Callable接口来编写异步执行任务");Thread.sleep(1000);return "返回 Second Callable 接口线程执行结果";};return callable;}private Callable<String> getFirstCallable() {Callable<String> callable  = () -> {System.out.println("实现 First Callable接口来编写异步执行任务");Thread.sleep(1000);return "返回 First Callable 接口线程执行结果";};return callable;}@Overridepublic void setApplicationContext(ApplicationContext applicationContext) throws BeansException {this.applicationContext = applicationContext;}@Overridepublic void onApplicationEvent(ContextStoppedEvent contextStoppedEvent) {try {THREAD_POOL_EXECUTOR.shutdown();} catch (Exception e) {log.error("停止线程池失败", e);}}
}

5. 利用CompleteFutable实现多线程异步任务执行

在 Java 8 中, 新增加了一个包含 50 个方法左右的类: CompletableFuture， 提供了非常强大的Future 的扩展功能， 可以帮助我们简化异步编程的复杂性， 提供了函数式编程的能力， 可以通过回调的方式处理计算结果， 并且提供了转换和组合 CompletableFuture 的方法。CompletableFuture 类实现了 Future 接口， 所以你还是可以像以前一样通过get方法阻塞或者轮询的方式获得结果， 但是这种方式不推荐使用。

CompletableFuture 和 FutureTask 同属于 Future 接口的实现类， 都可以获取线程的执行结果。

CompletableFuture 提供了四个静态方法来创建一个异步操作 ：

runXxxx 都是没有返回结果的， supplyXxx 都是可以获取返回结果的，可以传入自定义的线程池， 否则就用默认的线程池 。

//子任务包装一个Runnable实例，并调用ForkJoinPool.commonPool()线程池来执行
public static CompletableFuture<Void> runAsync(Runnable runnable)//子任务包装一个Runnable实例，并调用指定的executor线程池来执行
public static CompletableFuture<Void> runAsync(Runnable runnable, Executor executor)//子任务包装一个Supplier实例，并调用ForkJoinPool.commonPool()线程池来执行
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier)//子任务包装一个Supplier实例，并使用指定的executor线程池来执行
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier, Executor executor)

@Service
@Slf4j
public class BeanLoadService {public void getInfo() {List<CompletableFuture<String>> futures = new ArrayList<>();CompletableFuture<String> firstFuture = CompletableFuture.supplyAsync(() -> {try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}return "返回 First Callable 接口线程执行结果";});futures.add(firstFuture);CompletableFuture<String> secondFuture = CompletableFuture.supplyAsync(() -> {try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}return "返回 Second Callable 接口线程执行结果";});futures.add(secondFuture);CompletableFuture<String> thirdFuture = CompletableFuture.supplyAsync(() -> {try {Thread.sleep(1000);} catch (InterruptedException e) {e.printStackTrace();}return "返回 Third Callable 接口线程执行结果";});futures.add(thirdFuture);try {futures.forEach(future -> {try {System.out.println(future.get());} catch (Exception e) {log.warn("incident delay data,future get warn", e);}});} catch (Exception e) {log.warn("incident delay data warn ", e);}}
}