- 1.概述
- 2.为什么引入CompletableFuture
- 3.功能
- 4.源码追踪
- 4.总结
1.概述
CompletableFuture是jdk1.8引入的实现类。扩展了Future和CompletionStage，是一个可以在任务完成阶段触发一些操作Future。简单的来讲就是可以实现异步回调。
基于 Spring Boot + MyBatis Plus + Vue & Element 实现的后台管理系统 + 用户小程序，支持 RBAC 动态权限、多租户、数据权限、工作流、三方登录、支付、短信、商城等功能。
项目地址：https://github.com/YunaiV/ruoyi-vue-pro
2.为什么引入CompletableFuture
对于jdk1.5的Future，虽然提供了异步处理任务的能力，但是获取结果的方式很不优雅，还是需要通过阻塞（或者轮训）的方式。如何避免阻塞呢？其实就是注册回调。
业界结合观察者模式实现异步回调。也就是当任务执行完成后去通知观察者。比如Netty的ChannelFuture，可以通过注册监听实现异步结果的处理。
Netty的ChannelFuture
public Promise<V> addListener(GenericFutureListener<? extends Future<? super V>> listener) {
checkNotNull(listener, "listener");
synchronized (this) {
addListener0(listener);
}
if (isDone()) {
notifyListeners();
}
return this;
}
private boolean setValue0(Object objResult) {
if (RESULT_UPDATER.compareAndSet(this, null, objResult) ||
RESULT_UPDATER.compareAndSet(this, UNCANCELLABLE, objResult)) {
if (checkNotifyWaiters()) {
notifyListeners();
}
return true;
}
return false;
}
通过addListener方法注册监听。如果任务完成，会调用notifyListeners通知。
CompletableFuture通过扩展Future，引入函数式编程，通过回调的方式去处理结果。
基于微服务的思想，构建在 B2C 电商场景下的项目实战。核心技术栈，是 Spring Boot + Dubbo 。未来，会重构成 Spring Cloud Alibaba 。
项目地址：https://github.com/YunaiV/onemall
3.功能
CompletableFuture的功能主要体现在他的CompletionStage。
可以实现如下等功能
- 转换（thenCompose）
- 组合（thenCombine）
- 消费（thenAccept）
- 运行（thenRun）。
- 带返回的消费（thenApply）
消费和运行的区别：
消费使用执行结果。运行则只是运行特定任务。具体其他功能大家可以根据需求自行查看。
CompletableFuture借助CompletionStage的方法可以实现链式调用。并且可以选择同步或者异步两种方式。
这里举个简单的例子来体验一下他的功能。
public static void thenApply() {
ExecutorService executorService = Executors.newFixedThreadPool(2);
CompletableFuture cf = CompletableFuture.supplyAsync(() -> {
try {
// Thread.sleep(2000);
} catch (Exception e) {
e.printStackTrace();
}
System.out.println("supplyAsync " + Thread.currentThread().getName());
return "hello";
}, executorService).thenApplyAsync(s -> {
System.out.println(s + "world");
return "hhh";
}, executorService);
cf.thenRunAsync(() -> {
System.out.println("ddddd");
});
cf.thenRun(() -> {
System.out.println("ddddsd");
});
cf.thenRun(() -> {
System.out.println(Thread.currentThread());
System.out.println("dddaewdd");
});
}
执行结果
supplyAsync pool-1-thread-1
helloworld
ddddd
ddddsd
Thread[main,5,main]
dddaewdd
根据结果我们可以看到会有序执行对应任务。
注意：
如果是同步执行cf.thenRun。他的执行线程可能main线程，也可能是执行源任务的线程。如果执行源任务的线程在main调用之前执行完了任务。那么cf.thenRun方法会由main线程调用。
这里说明一下，如果是同一任务的依赖任务有多个：
- 如果这些依赖任务都是同步执行。那么假如这些任务被当前调用线程（main）执行，则是有序执行，假如被执行源任务的线程执行，那么会是倒序执行。因为内部任务数据结构为LIFO。
- 如果这些依赖任务都是异步执行，那么他会通过异步线程池去执行任务。不能保证任务的执行顺序。
上面的结论是通过阅读源代码得到的。下面我们深入源代码。
4.源码追踪
创建CompletableFuture
创建的方法有很多，甚至可以直接new一个。我们来看一下supplyAsync异步创建的方法。
public static <U> CompletableFuture<U> supplyAsync(Supplier<U> supplier,
Executor executor) {
return asyncSupplyStage(screenExecutor(executor), supplier);
}
static Executor screenExecutor(Executor e) {
if (!useCommonPool && e == ForkJoinPool.commonPool())
return asyncPool;
if (e == null) throw new NullPointerException();
return e;
}
入参Supplier，带返回值的函数。如果是异步方法，并且传递了执行器，那么会使用传入的执行器去执行任务。否则采用公共的ForkJoin并行线程池，如果不支持并行，新建一个线程去执行。
这里我们需要注意ForkJoin是通过守护线程去执行任务的。所以必须有非守护线程的存在才行。
asyncSupplyStage方法
static <U> CompletableFuture<U> asyncSupplyStage(Executor e,
Supplier<U> f) {
if (f == null) throw new NullPointerException();
CompletableFuture<U> d = new CompletableFuture<U>();
e.execute(new AsyncSupply<U>(d, f));
return d;
}
这里会创建一个用于返回的CompletableFuture。
然后构造一个AsyncSupply，并将创建的CompletableFuture作为构造参数传入。
那么，任务的执行完全依赖AsyncSupply。
AsyncSupply#run
public void run() {
CompletableFuture<T> d; Supplier<T> f;
if ((d = dep) != null && (f = fn) != null) {
dep = null; fn = null;
if (d.result == null) {
try {
d.completeValue(f.get());
} catch (Throwable ex) {
d.completeThrowable(ex);
}
}
d.postComplete();
}
}
- 该方法会调用Supplier的get方法。并将结果设置到CompletableFuture中。我们应该清楚这些操作都是在异步线程中调用的。
- d.postComplete方法就是通知任务执行完成。触发后续依赖任务的执行，也就是实现CompletionStage的关键点。
在看postComplete方法之前我们先来看一下创建依赖任务的逻辑。
thenAcceptAsync方法
public CompletableFuture<Void> thenAcceptAsync(Consumer<? super T> action) {
return uniAcceptStage(asyncPool, action);
}
private CompletableFuture<Void> uniAcceptStage(Executor e,
Consumer<? super T> f) {
if (f == null) throw new NullPointerException();
CompletableFuture<Void> d = new CompletableFuture<Void>();
if (e != null || !d.uniAccept(this, f, null)) {
# 1
UniAccept<T> c = new UniAccept<T>(e, d, this, f);
push(c);
c.tryFire(SYNC);
}
return d;
}
上面提到过。thenAcceptAsync是用来消费CompletableFuture的。该方法调用uniAcceptStage。
uniAcceptStage逻辑：
- 构造一个CompletableFuture，主要是为了链式调用。
- 如果为异步任务，直接返回。因为源任务结束后会触发异步线程执行对应逻辑。
- 如果为同步任务（e==null），会调用d.uniAccept方法。这个方法在这里逻辑：如果源任务完成，调用f，返回true。否则进入if代码块（Mark 1）。
- 如果是异步任务直接进入if（Mark 1）。
Mark1逻辑：
- 构造一个UniAccept，将其push入栈。这里通过CAS实现乐观锁实现。
- 调用c.tryFire方法。
final CompletableFuture<Void> tryFire(int mode) {
CompletableFuture<Void> d; CompletableFuture<T> a;
if ((d = dep) == null ||
!d.uniAccept(a = src, fn, mode > 0 ? null : this))
return null;
dep = null; src = null; fn = null;
return d.postFire(a, mode);
}
- 会调用d.uniAccept方法。其实该方法判断源任务是否完成，如果完成则执行依赖任务，否则返回false。
- 如果依赖任务已经执行，调用d.postFire，主要就是Fire的后续处理。根据不同模式逻辑不同。
这里简单说一下，其实mode有同步异步，和迭代。迭代为了避免无限递归。
这里强调一下d.uniAccept方法的第三个参数。
如果是异步调用（mode>0），传入null。否则传入this。
区别看下面代码。c不为null会调用c.claim方法。
try {
if (c != null && !c.claim())
return false;
@SuppressWarnings("unchecked") S s = (S) r;
f.accept(s);
completeNull();
} catch (Throwable ex) {
completeThrowable(ex);
}
final boolean claim() {
Executor e = executor;
if (compareAndSetForkJoinTaskTag((short)0, (short)1)) {
if (e == null)
return true;
executor = null; // disable
e.execute(this);
}
return false;
}
claim方法是逻辑：
- 如果异步线程为null。说明同步，那么直接返回true。最后上层函数会调用f.accept(s)同步执行任务。
- 如果异步线程不为null，那么使用异步线程去执行this。
this的run任务如下。也就是在异步线程同步调用tryFire方法。达到其被异步线程执行的目的。
public final void run(){
tryFire(ASYNC);
}
看完上面的逻辑，我们基本理解依赖任务的逻辑。
其实就是先判断源任务是否完成，如果完成，直接在对应线程执行以来任务（如果是同步，则在当前线程处理，否则在异步线程处理）
如果任务没有完成，直接返回，因为等任务完成之后会通过postComplete去触发调用依赖任务。
postComplete方法
final void postComplete() {
/*
* On each step, variable f holds current dependents to pop
* and run. It is extended along only one path at a time,
* pushing others to avoid unbounded recursion.
*/
CompletableFuture<?> f = this; Completion h;
while ((h = f.stack) != null ||
(f != this && (h = (f = this).stack) != null)) {
CompletableFuture<?> d; Completion t;
if (f.casStack(h, t = h.next)) {
if (t != null) {
if (f != this) {
pushStack(h);
continue;
}
h.next = null; // detach
}
f = (d = h.tryFire(NESTED)) == null ? this : d;
}
}
}
在源任务完成之后会调用。
其实逻辑很简单，就是迭代堆栈的依赖任务。调用h.tryFire方法。NESTED就是为了避免递归死循环。因为FirePost会调用postComplete。如果是NESTED，则不调用。
堆栈的内容其实就是在依赖任务创建的时候加入进去的。上面我们已经提到过。
4.总结
基本上述源码已经分析了逻辑。
因为涉及异步等操作，我们需要理一下（这里针对全异步任务）：
- 创建CompletableFuture成功之后会通过异步线程去执行对应任务。
- 如果CompletableFuture还有依赖任务（异步），会将任务加入到CompletableFuture的堆栈保存起来。以供后续完成后执行依赖任务。
当然，创建依赖任务并不只是将其加入堆栈。如果源任务在创建依赖任务的时候已经执行完成，那么当前线程会触发依赖任务的异步线程直接处理依赖任务。并且会告诉堆栈其他的依赖任务源任务已经完成。
主要是考虑代码的复用。所以逻辑相对难理解。
postComplete方法会被源任务线程执行完源任务后调用。同样也可能被依赖任务线程后调用。
执行依赖任务的方法主要就是靠tryFire方法。因为这个方法可能会被多种不同类型线程触发，所以逻辑也绕一点。（其他依赖任务线程、源任务线程、当前依赖任务线程）
- 如果是当前依赖任务线程，那么会执行依赖任务，并且会通知其他依赖任务。
- 如果是源任务线程，和其他依赖任务线程，则将任务转换给依赖线程去执行。不需要通知其他依赖任务，避免死递归。
不得不说Doug Lea的编码，真的是艺术。代码的复用性全体现在逻辑上了。
面试官：为什么数据库连接很消耗资源，资源都消耗在哪里？
2022-04-11 19:19·小程序建站
本文主要想探究一下连接数据库的细节，尤其是在Web应用中要使用数据库来连接池，以免每次发送一次请求就重新建立一次连接。对于这个问题，答案都是一致的，建立数据库连接很耗时，但是这个耗时是都多少呢，又是分别在哪些方面产生的耗时呢？
本文以连接MySQL数据库为例，因为MySQL数据库是开源的，其通信协议是公开的，所以我们能够详细分析建立连接的整个过程。
?
在本文中，消耗资源的分析主要集中在网络上，当然，资源也包括内存、CPU等计算资源，使用的编程语言是Java，但是不排除编程语言也会有一定的影响。
?
首先先看一下连接数据库的Java代码，如下：
Class.forName("com.mysql.jdbc.Driver");
String name = "xttblog2";
String password = "123456";
String url = "jdbc:mysql://172.16.100.131:3306/xttblog2";
Connection conn = DriverManager.getConnection(url, name, password);
// 之后程序终止，连接被强制关闭
然后通过「Wireshark」 分析整个连接的建立过程，如下：
Wireshark抓包
在上图中显示的连接过程中，可以看出MySQL的通信协议是基于TCP传输协议的，而且该协议是二进制协议，不是类似于HTTP的文本协议，其中建立连接的过程具体如下：
- 第1步：建立TCP连接，通过三次握手实现；
- 第2步：服务器发送给客户端「握手信息」 ，客户端响应该握手消息；
- 第3步：客户端「发送认证包」 ，用于用户验证，验证成功后，服务器返回OK响应，之后开始执行命令；
用户验证成功之后，会进行一些连接变量的设置，比如字符集、是否自动提交事务等，其间会有多次数据的交互。完成了这些步骤后，才会执行真正的数据查询和更新等操作。
在本文的测试中，只用了5行代码来建立连接，但是并没有通过该连接去执行任何操作，所以在程序执行完毕之后，连接不是通过Connection.close()关闭的，而是由于程序执行完毕，导致进程终止，造成与数据库的连接异常关闭，所以最后会出现TCP的RST报文。在这个最简单的代码中，没有设置任何额外的连接属性，所以在设置属性上占用的时间可以认为是最少的（其实，虽然我们没有设置任何属性，但是驱动仍然设置了字符集、事务自动提交等，这取决于具体的驱动实现），所以整个连接所使用的时间可以认为是最少的。但从统计信息中可以看出，在不包括最后TCP的RST 报文时（因为该报文不需要服务器返回任何响应），但是其中仍需在客户端和服务器之间进行往返「7」 次，「也就是说完成一次连接，可以认为，数据在客户端和服务器之间需要至少往返7次」 ，从时间上来看，从开始TCP的三次握手，到最终连接强制断开为止（不包括最后的RST报文），总共花费了：
10.416042 - 10.190799 = 0.225243s = **225.243ms**！！！
这意味着，建立一次数据库连接需要225ms，而这还是还可以认为是最少的，当然「花费的时间可能受到网络状况、数据库服务器性能以及应用代码是否高效的影响」 ，但是这里只是一个最简单的例子，已经足够说明问题了！
由于上面是程序异常终止了，但是在正常的应用程序中，连接的关闭一般都是通过Connection.close()完成的，代码如下：
Class.forName("com.mysql.jdbc.Driver");
String name = "shine_user";
String password = "123";
String url = "jdbc:mysql://172.16.100.131:3306/clever_mg_test";
Connection conn = DriverManager.getConnection(url, name, password);
conn.close();
这样的话，情况发生了变化，主要体现在与数据库连接的断开，如下图：
网络抓包
- 第1步：此时处于MySQL通信协议阶段，客户端发送关闭连接请求，而且不用等待服务端的响应；
- 第2步：TCP断开连接，4次挥手完成连接断开；
这里是完整地完成了从数据库连接的建立到关闭，整个过程花费了：
747.284311 - 747.100954 = 0.183357s = 183.357ms
这里可能也有网络状况的影响，比上述的225ms少了，但是也几乎达到了200ms的级别。
那么问题来了，想象一下这个场景，对于一个日活2万的网站来说，假设每个用户只会发送5个请求，那么一天就是10万个请求，对于建立数据库连接，我们保守一点计算为150ms好了，那么一天当中花费在建立数据库连接的时间有（还不包括执行查询和更新操作）：
100000 * 150ms = 15000000ms = 15000s = 250min = 4.17h
也就说每天花费在建立数据库连接上的时间已经达到「4个小时」 ，所以说数据库连接池是必须的嘛，而且当日活增加时，单单使用数据库连接池也不能完全保证你的服务能够正常运行，还需要考虑其他的解决方案：
- 缓存
- SQL的预编译
- 负载均衡
- ……
当然这不是本文的主要内容，「本文想要阐述的核心思想只有一个，数据库连接真的很耗时，所以不要频繁的建立连接」 。
本人花费2个月时间，整理了一套JAVA开发技术资料，内容涵盖java基础，分布式、微服务等主流技术资料，包含大厂面经，学习笔记、源码讲义、项目实战、讲解视频。
希望可以帮助一些想通过自学提升能力的朋友，领取资料，扫码关注一下
记得转发+关注+私信
私信回复【2022面试资料】
领取更多学习资料