Java通过 CompletableFuture 实现异步多线程优化请求处理速度(等待多个线程执行完成后再继续操作)
零、背景
我们在写后端请求的时候, 可能涉及多次 SQL 执行(或其他操作), 当这些请求相互不关联, 在顺序执行时就浪费了时间, 这些不需要先后顺序的操作可以通过多线程进行同时执行, 来加速整个逻辑的执行速度.
既然有了目标和大致思路, 如果有做过前端的小伙伴应该能想起来 Js 里面有个 Promise.all
来解决这个问题, 在 Java 里也有类似功能的类 CompletableFuture
, 它可以实现多线程和线程阻塞, 这样能够保证等待多个线程执行完成后再继续操作.
一、CompletableFuture 是什么
首先我们先了解一下 CompletableFuture
是干什么, 接下来我们通过简单的示例来介绍他的作用.
long startTime = System.currentTimeMillis();
//生成几个任务
List<CompletableFuture<String>> futureList = new ArrayList<>();
futureList.add(CompletableFuture.supplyAsync(()->{
sleep(4000);
System.out.println("任务1 完成");
return "任务1的数据";
}));
futureList.add(CompletableFuture.supplyAsync(()->{
sleep(2000);
System.out.println("任务2 完成");
return "任务2的数据";
}));
futureList.add(CompletableFuture.supplyAsync(()->{
sleep(3000);
System.out.println("任务3 完成");
return "任务3的数据";
}));
//完成任务
CompletableFuture<Void> allTask = CompletableFuture.allOf(futureList.toArray(new CompletableFuture[0]))
.whenComplete((t, e) -> {
System.out.println("所有任务都完成了, 返回结果集: "
+ futureList.stream().map(CompletableFuture::join).collect(Collectors.joining(",")));
});
// 阻塞主线程
allTask.join();
System.out.println("main end, cost: " + (System.currentTimeMillis() - startTime));
执行结果
任务2 完成
任务3 完成
任务1 完成
所有任务都完成了, 返回结果集: 任务1的数据,任务2的数据,任务3的数据
main end, cost: 4032
结果分析: 我们需要执行3个任务, 3个任务同时执行, 互不影响
- 其中任务2耗时最短,提前打印完成
- 其次是任务3
- 最后是执行1完成
- 当所有任务完成后, 触发
whenComplete
方法, 打印任务的返回结果 - 最后打印总耗时为 4.032s
- 结论: 多线程执行后, 耗时取决于最耗时的操作, 而单线程是所有操作耗时之和
二、封装工具类
经过上面的测试, 通过 CompletableFuture
已经能够实现我们的预想, 为了操作方便, 我们将封装起来, 便于统一管理
package org.yelog.java.usage.concurrent;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CompletableFuture;
import java.util.function.Consumer;
import java.util.function.Function;
import java.util.function.Predicate;
/**
* 执行并发任务
*
* @author yangyj13
* @date 11/7/22 9:49 PM
*/
public class MultiTask<T> {
private List<CompletableFuture<T>> futureList;
/**
* 添加待执行的任务
*
* @param completableFuture 任务
* @return 当前对象
*/
public MultiTask<T> addTask(CompletableFuture<T> completableFuture) {
if (futureList == null) {
futureList = new ArrayList<>();
}
futureList.add(completableFuture);
return this;
}
/**
* 添加待执行的任务(无返回)
*
* @param task 任务
* @return 当前对象
*/
public MultiTask<T> addTask(Consumer<T> task) {
addTask(CompletableFuture.supplyAsync(() -> {
task.accept(null);
return null;
}));
return this;
}
/**
* 添加待执行的任务(有返回)
*
* @param task 任务
* @return 当前对象
*/
public MultiTask<T> addTask(Function<Object, T> task) {
addTask(CompletableFuture.supplyAsync(() -> task.apply(null)));
return this;
}
/**
* 开始执行任务
*
* @param callback 当所有任务都完成后触发的回调方法
* @param waitTaskExecuteComplete 是否阻塞主线程
*/
private void execute(Consumer<List<T>> callback, Boolean waitTaskExecuteComplete) {
CompletableFuture<Void> allFuture = CompletableFuture.allOf(futureList.toArray(new CompletableFuture[0]))
.whenComplete((t, e) -> {
if (callback != null) {
List<T> objectList = new ArrayList<>();
futureList.forEach((future) -> {
objectList.add(future.join());
});
callback.accept(objectList);
}
});
if (callback != null || waitTaskExecuteComplete == null || waitTaskExecuteComplete) {
allFuture.join();
}
}
/**
* 开始执行任务
* 等待所有任务完成(阻塞主线程)
*/
public void execute() {
execute(null, true);
}
/**
* 开始执行任务
*
* @param waitTaskExecuteComplete 是否阻塞主线程
*/
public void execute(Boolean waitTaskExecuteComplete) {
execute(null, waitTaskExecuteComplete);
}
/**
* 开始执行任务
*
* @param callback 当所有任务都完成后触发的回调方法
*/
public void execute(Consumer<List<T>> callback) {
execute(callback, true);
}
}
那么上一步我们测试的流程转换成工具类后如下
long startTime = System.currentTimeMillis();
MultiTask<String> multiTask = new MultiTask<>();
multiTask.addTask(t -> {
sleep(1000);
System.out.println("任务1 完成");
}).addTask(t -> {
sleep(3000);
System.out.println("任务2 完成");
}).addTask(CompletableFuture.supplyAsync(()->{
sleep(2000);
System.out.println("任务3 完成");
return "任务3的数据";
})).execute(resultList->{
System.out.println("all complete: " + resultList);
});
System.out.println("main end, cost: " + (System.currentTimeMillis() - startTime));
三、应用到实际的效果
执行两次数据库的操作如下
public interface TestMapper {
@Select("select count(*) from test_user where score < 1000 and user_id = #{userId}")
int countScoreLess1000(Integer userId);
@Select("select count(1) from test_log where success = true and user_id = #{userId}")
int countSuccess(Integer userId);
}
调用方法:
long start = System.currentTimeMillis();
testMapper.countScoreLess1000(userId);
long countScoreLess1000End = System.currentTimeMillis();
log.info("countScoreLess1000 cost: " + (countScoreLess1000End - start));
testMapper.countSuccess(userId);
long countSuccessEnd = System.currentTimeMillis();
log.info("countSuccess cost: " + (countSuccessEnd - countScoreLess1000End));
log.info("all cost: " + (countSuccessEnd - start));
顺序执行的平均时间如下
countScoreLess1000 cost: 368
countSuccess cost: 404
all cost: 772
当我们应用的上面的工具类后的调用方法
MultiTask multiTask = new MultiTask<>();
multiTask.addTask(t -> {
testMapper.countScoreLess1000(userId);
log.info("countScoreLess1000 cost: " + (System.currentTimeMillis() - start));
}).addTask(t -> {
testMapper.countSuccess(userId);
log.info("countSuccess cost: " + (System.currentTimeMillis() - start));
}).execute();
log.info("all cost: " + (System.currentTimeMillis() - start));
效果如下
countScoreLess1000 cost: 433
countSuccess cost: 463
all cost: 464
可以看到各子任务执行时长是差不多的, 但是总耗时使用多线程后有了明显下降
四、总结
通过使用 CompletableFuture
实现多线程阻塞执行后, 大幅降低这类请求, 并且当可以异步执行的子任务越多, 效果越明显.
本文来自 叶落阁
近期,123云盘开始限制,必须注册帐号且登录后才能下载,且限制每天下载流量1GB,下载 123云盘免流量破解工具
版权声明:
小编:牛A与牛C之间
链接:https://www.wuleba.com/2257.html
来源:吾乐吧软件站
本站资源仅供个人学习交流,请于下载后 24 小时内删除,不允许用于商业用途,否则法律问题自行承担。
下载说明:请不要相信网盘那边的广告,不要下载APP,普通下载就完事了!
共有 0 条评论