JAVA常用的异步处理方法总结

前言

在java项目开发过程中经常会遇到比较耗时的任务，通常是将这些任务做成异步操作，在java中实现异步操作有很多方法，本文主要总结一些常用的处理方法。为了简化，我们就拿一个实际的案例，再用每种方法去实现，对比看看这些方法有什么优缺点。

具体案例：
在C://img/url.txt中有1000个图片URL，我们需要将这些图片下载到C://img/download目录下。
需要将每张图片耗时累加起来，输出最后的时间

img.txt数据格式

https://a.com/1.jpg
https://a.com/2.jpg
https://a.com/3.jpg
...

计算Demo

比如 下载1.jpg 耗时1ms ,下载 2.jpg耗时2ms，…下载 n.jpg耗时 n ms
最终我们总耗时 1+2+3+…1000=500500ms

公共方法

为了方便测试我们先定义一个DownloadImg下载接口

public interface DownloadImg {/**** @param urls 需要下载图片的url* @return  所有图片下载总耗时ms*/Long download(List<String> urls) throws Exception ;/**** @param url 下载单个图片的URL* @return 下载单张图片耗时 ms*/Long download(String url) throws Exception;/*** 关闭线程池*/void  shutdown() throws InterruptedException;/***  提交任务* @param task* @return*/Future<?> submit(Runnable task);
}

为了简化下载、创建线程池、时间累加等操作，我们抽象一些共用方法：

public abstract class AbstractDownloadImg implements DownloadImg {//总耗时public  AtomicLong sumTime=new AtomicLong(0L);public ExecutorService executorService = Executors.newFixedThreadPool(10);private AtomicInteger i = new AtomicInteger(1);public Long download(String url) {try {Long startTime = System.currentTimeMillis();FileUtils.copyURLToFile(new URL(url), new File(String.format("C:\img\download\%s.jpg", i.getAndIncrement())));Long castTime = System.currentTimeMillis() - startTime;System.out.println(Thread.currentThread().getName() + " download :" + url + " success, cast :" + castTime + " ms");return castTime;} catch (Exception e) {e.printStackTrace();return 0L;}}public void shutdown() throws InterruptedException {Thread.sleep(1000L);executorService.shutdown();}public Future<?> submit(Runnable task) {return executorService.submit(task);}
}

具体实现

一. 使用Future任务多线程下载

这种方法很是自然而然能想到，文件中有1000个图片，单个线程依次去下载太慢了，于是我们可以把1000张图片分成10个子任务，每个子任务去下载100张图片，子任务中把这100张图片耗时加起来，然后再把这10个子任务的耗时相加就是总时长了。

public class MultiThreadDownload extends AbstractDownloadImg implements DownloadImg {@Overridepublic Long download(List<String> imgUrls) throws Exception {//每100个一组List<List<String>> urls = Lists.partition(imgUrls, 100);//每个线程下载100张图片耗时返回结果List<Future<Long>> futures = new ArrayList<>();//分成10个线程，每个线程下载100个urls.forEach(subUrls->{FutureTask<Long> future= new FutureTask<>(() -> subUrls.stream().map(this::download).mapToLong(x->x).sum());//反回结果添加到futures中futures.add(future);//提交到线程池中submit(future);});//每线程耗时时间累加for(Future<Long> f:futures){sumTime.getAndAdd(f.get());}return sumTime.get();}
}

优点 ：比较简单，大部人第一眼能想到的方法

缺点 ：会产生水桶效应。如果前9个线程下载的都是小图片，很快下载完成了，第10个线程全是大图片，当最后9个线都空着时，第10个线程任务可能还在等待

二. CompletableFuture

上面的方法，是我们自己写Future，然后拿到返回值相加，在JUC包下面有个CompletableFuture，我们可以直接拿来用。

public class CompletableFutureDownload extends AbstractDownloadImg implements DownloadImg {@Overridepublic Long download(List<String> imgUrls) {//创建10个CompletableFutureCompletableFuture<Long>[] completableFutures = new CompletableFuture[imgUrls.size()];for (int i = 0; i < imgUrls.size(); i++) {String url = imgUrls.get(i);completableFutures[i] = CompletableFuture.supplyAsync(() -> download(url), executorService).whenComplete((k, v) -> sumTime.getAndAdd(k));}//所有任务合成一个CompletableFutureCompletableFuture allFuture = CompletableFuture.allOf(completableFutures).whenComplete((k, v) -> {System.out.println("all future complete cast: {} " + sumTime.get() + "ms");});//等待所有任务完成allFuture.join();return sumTime.get();}
}

三. 使用CountDownLatch

使用Futrue获取线程池返回结果还是有点麻烦的，在JUC包中有个CountDownLatch(倒计数门闩),使用这个实现代码就简化很多了，我们只需要把每张图片下载耗时累加起来，最后等待所有任务完成就OK了。

public class CountDownLatchDownload extends AbstractDownloadImg implements DownloadImg {@Overridepublic Long download(List<String> imgUrls) throws Exception {//门栓计数次CountDownLatch countDownLatch = new CountDownLatch(imgUrls.size());//总耗时AtomicLong sumTime = new AtomicLong(0L);for (String url : imgUrls) {submit(() -> {try {sumTime.getAndAdd(download(url));} finally {countDownLatch.countDown();}});}//等待所有任务结束countDownLatch.await();return sumTime.get();}
}

四. 使用lambda中的parallelStream

有了上面倒计数门闩，那我们自然可以想到JAVA8 lambda中的parallelStream了，用了parallelStream上面的代码又可以简化了。

public class ParallelStreamDownload extends AbstractDownloadImg implements DownloadImg {@Overridepublic Long download(List<String> imgUrls) throws Exception {return imgUrls.parallelStream().mapToLong(this::download).sum();}
}

需要注意的地parallelStream底层实现是使用的fork join,默认线程数是CPU核数，而且是全局共用一个线程池的，这点很重要，如果不指定线程池，项目别处使用了parallelStream，可能影响你你当前这处代码的执行速度。当然我们可以设置默认线程数和指定线程池。

//设置parallelStream线程数量为20个
System.setProperty("java.util.concurrent.ForkJoinPool.common.parallelism", "20");

指定ForkJoinPool的parallelStream

public class ParallelStreamDownload extends AbstractDownloadImg implements DownloadImg {@Overridepublic Long download(List<String> imgUrls) throws Exception {ForkJoinPool forkJoinPool = new ForkJoinPool(20);return forkJoinPool.submit(() -> imgUrls.parallelStream().mapToLong(this::download).sum()).get();}
}

优点 ：代码非常简洁

缺点 ：隐藏了很多细节，使用不当可能导致不可预估的后果，如果不了解内部原理，你都不知道为什么你的代码卡住了

五. 使用Fork/Join

parallelStream底层就是用的Fork/Join来实现的，所以我们也可以自己用Fork/Join来实现。

public class ForkJoinDownload extends AbstractDownloadImg implements DownloadImg {static class DownloadJoinTask extends RecursiveTask<Long> {//需要下载的URLprivate List<String> urls;//子任务最多条数private Integer MAX_TASK_COUNT = 100;private DownloadImg downloadImg;public DownloadJoinTask(List<String> urls, DownloadImg downloadImg) {this.urls = urls;this.downloadImg = downloadImg;}@Overrideprotected Long compute() {//当前任务<=100个，执行下载操作if (urls.size() <= MAX_TASK_COUNT) {return urls.stream().map(x -> {try {return downloadImg.download(x);} catch (Exception e) {e.printStackTrace();return 0L;}}).mapToLong(x -> x).sum();} else {//当前前任务拆分成两个任务ForkJoinDownload.DownloadJoinTask leftTask = new ForkJoinDownload.DownloadJoinTask(urls.subList(0, urls.size() / 2), downloadImg);ForkJoinDownload.DownloadJoinTask rightTask = new ForkJoinDownload.DownloadJoinTask(urls.subList(urls.size() / 2, urls.size()), downloadImg);//提交子任务invokeAll(leftTask, rightTask);return leftTask.join() + rightTask.join();}}}@Overridepublic Long download(List<String> imgUrls) throws Exception {ForkJoinPool forkJoinPool = new ForkJoinPool(20);DownloadJoinTask downloadJoinTask = new DownloadJoinTask(imgUrls, this);ForkJoinTask<Long> taskFuture = forkJoinPool.submit(downloadJoinTask);sumTime.addAndGet(taskFuture.get());return sumTime.get();}
}

优点 ：工作窃取算法,不会产生水桶效应

缺点 ：需要正确理解Fork/Join模型的任务执行逻辑才能写出好代码，有一定的门槛

测试代码

public static void main(String[] args) throws Exception {List<String> urls = FileUtils.readLines(new File("C:\img\url.txt"), Charset.defaultCharset());//DownloadImg downloadImg=new MultiThreadDownload();//DownloadImg downloadImg=new CountDownLatchDownload();//DownloadImg downloadImg=new ParallelStreamDownload();//DownloadImg downloadImg=new ForkJoinDownload();DownloadImg downloadImg = new CompletableFutureDownload();System.out.println("download all url cast: " + downloadImg.download(urls) + " ms");downloadImg.shutdown();
}

总结

本文主要通过下载图片这个具体的案例，介绍JAVA中5种常用的方法如何异步处理比较耗时的任务，并对比了优缺点，希望在项目过遇到类似的需求，可以帮助你找到合适的方法。