JAVA常用的异步处理方法总结
站长
· 阅读数 13
前言
在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 {
@Override
public 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 {
@Override
public Long download(List<String> imgUrls) {
//创建10个CompletableFuture
CompletableFuture<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));
}
//所有任务合成一个CompletableFuture
CompletableFuture 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 {
@Override
public 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 {
@Override
public 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 {
@Override
public 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> {
//需要下载的URL
private 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;
}
@Override
protected 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();
}
}
}
@Override
public 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种常用的方法如何异步处理比较耗时的任务，并对比了优缺点，希望在项目过遇到类似的需求，可以帮助你找到合适的方法。
转载自:https://juejin.cn/post/7227657672204009529