theme: orange
本文为稀土掘金技术社区首发签约文章，30天内禁止转载，30天后未获授权禁止转载，侵权必究！
在实际工作场景中，我们往往要面对各种各样的优化问题：
- 接口 RTS 时间太长？异步优化！
- 批处理任务执行时间太长？并行优化！
- CPU 利用率太低？多线程优化！
在各种各样的场景优化中，我们最常用到的技术是多线程 + 线程池的方案。
即：起一个线程池，进行异步处理 or 对批量任务进行多线程并行处理。
优化的本质是利用更少的资源执行相同或者更多的任务，但是哪怕你在代码中多次使用了线程池，大概率你的应用实例 CPU 利用率依然是不温不火的。
因为像我们这种性能瓶颈在 IO 的 web 应用，只依靠多线程技术并不能非常好的提高你的资源使用率，反而有可能线程变多了，IO 瓶颈还在，大量的线程只是在徒增上下文切换。
1. 线程池任务示例
这里我将举一个小小例子的来说明这点：
public void completionServiceTest() throws InterruptedException, ExecutionException {
// 创建线程池
ExecutorService executor = Executors.newFixedThreadPool(100);
// 创建 CompletionService
CompletionService<String> completionService = new ExecutorCompletionService<>(executor);
// 创建计数器
AtomicInteger count = new AtomicInteger(0);
for (int i = 0; i < 10000; i++) {
completionService.submit(() -> {
count.incrementAndGet();
String json = HttpUtil
.get("https://dict-mobile.iciba.com/interface/index.php?c=word&m=getsuggest&nums=10&is_need_mean=1&word=h");
return json;
});
}
completionService.take().get();
logger.info("count: " + count.get());
}
public static void main(String[] args) throws InterruptedException, ExecutionException {
ReactorTest test = new ReactorTest();
test.completionServiceTest();
}
这里我创建了一个 100 个线程的线程池，接着使用 CompletionService 来管理线程池中的任务，接着为这个线程池添加了一万个任务，每个任务就计数+1，接着发送一个请求，用来模拟网络 IO。
当线程池中的第一个任务结束时，主程序会退出（在 main 方法中调用，所以方法执行结束后，主程序会退出，第一个任务没有结束之前，会被 get 方法 pending 住），然后打印总共执行了多少个任务。
最终的打印结果会在 100 - 110 浮动，这个浮动其实可以忽略不计，因为count 计数是先 IO 一步的，count 增加了，不代表任务就结束了。
所以按照理论数值来说，100 个线程，其实可以同时执行的任务是 100（由于发送请求时依赖 http 客户端，所以如果你使用公用 http 连接池，结果可能会小于 100），那就出现了一个问题：
发送请求会阻塞 IO，那么在等待返回的过程中，线程此刻其实是空闲的，空闲的线程应该去执行其他任务的代码，这样才能提高 CPU 的利率，而不是傻傻的等待。
这一切是原因就是：线程执行过程中，依然是同步的，所以在遇到网络 IO 时，线程会被完全阻塞。
那么如何解决呢？
2. Reactor事件驱动带来的增益
聪明的同学可能从标题就猜出来我要讲 Reactor 模式(它是什么不再赘述)和其最强实现：Netty。
但是这次，有亿点点不一样，我们在使用 Netty 的过程中往往是作为服务端使用，在其实对于很多场景我们需要的是一个：实现了 Reactor 事件驱动的 Http 客户端（下面简称：响应式客户端）。
比如你的业务大量依赖第三方的接口，这其实很常见，因为第三方不光是外部系统，你的微服务的下游都可以算作你的第三方。
这时你如果依然使用 http 同步调用就会在 IO 时阻塞，而你如果使用一个响应式客户端就可以在网络 IO 时，去做其他任务中的耗费 CPU 的操作：数据转换、编解码和加密。
通过这种方式，就可以加快任务执行效率，比如你的任务分为两个步骤：
- 步骤一是编解码和数据加密。
- 步骤二是将组装好的数据通过 http 客户端发出去。
那么如果你使用线程池，同时发出去 100 个请求在进行 IO 等待时，其他已提交的任务中的步骤一并不会被执行，但是你使用响应式客户端却可以。
接下来来看两个比较常用的响应式客户端的例子：Vert.x 和 Spring Webflux，这俩都是基于 Netty的，因为 Netty 自己的客户端用起来比较复杂，才会有 Vert.x 这样的库～
Vert.x 目前属于 Eclipse 基金会，Eclipse这个名字估计大家比好久都没听过了，Vert.x 一开始只是一个响应式编程工具包，现在已经是一个构建于 JVM 上的响应式编程生态了，本次我们将使用到它的 Vert.x Web Client。
至于Spring Webflux，用过 Spring Gateway 的同学可能比较了解，它是 Spring Gateway 中的默认 Web 服务器，也是 Spring 家族自己开发的响应式编程 Web 服务器。
3. Vert.x & Spring WebFlux
先来引入一下依赖：
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-webflux</artifactId>
<version>3.2.0</version>
</dependency>
<dependency>
<groupId>io.vertx</groupId>
<artifactId>vertx-web-client</artifactId>
<version>4.5.7</version>
</dependency>
接下来是两个和上面线程池一样的示例，不同的是 Http 客户端不再使用 JDK 自带的，而是换成了这两个框架的：
package reactor;
import cn.hutool.http.HttpUtil;
import io.vertx.core.Vertx;
import io.vertx.ext.web.client.WebClient;
import io.vertx.ext.web.client.WebClientOptions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.http.client.reactive.ReactorClientHttpConnector;
import reactor.core.publisher.Flux;
import reactor.netty.http.client.HttpClient;
import reactor.netty.resources.ConnectionProvider;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicInteger;
public class ReactorTest {
private final Logger logger = LoggerFactory.getLogger(ReactorTest.class);
public void vertxTest() throws InterruptedException {
Vertx vertx = Vertx.vertx();
WebClientOptions options = new WebClientOptions()
.setMaxPoolSize(100)
.setConnectTimeout(5000)
.setIdleTimeout(60000)
.setKeepAlive(true);
WebClient client = WebClient.create(vertx, options);
AtomicInteger count = new AtomicInteger(0);
logger.info("START");
for (int i = 0; i < 1000; i++) {
count.incrementAndGet();
client
.get("suggest.taobao.com", "/sug?code=utf-8&q=%E5%8D%AB%E8%A1%A3&callback=cb")
.send()
.onSuccess(response -> {
logger.info("success count：" + count.get());
System.exit(0);
})
.onFailure(error -> {
logger.error(error.getLocalizedMessage());
logger.info("error count：" + count.get());
System.exit(0);
});
}
}
public void webFluxTest() {
HttpClient httpClient = HttpClient.create(ConnectionProvider.builder("webFlux").maxConnections(1000).build());
ReactorClientHttpConnector connector = new ReactorClientHttpConnector(httpClient);
org.springframework.web.reactive.function.client.WebClient client = org.springframework.web.reactive.function.client.WebClient.builder()
.clientConnector(connector)
.build();
AtomicInteger count = new AtomicInteger(0);
// 循环 0-1000 执行 1000 次 flatMap 中的代码
Flux.range(0, 1000)
.flatMap(i -> {
count.incrementAndGet();
return client.get()
.uri("https://suggest.taobao.com/sug?code=utf-8&q=%E5%8D%AB%E8%A1%A3&callback=cb")
.retrieve()
.bodyToMono(String.class);
}
)
.subscribe(
result -> {
logger.info(result);
logger.info("success count：" + count.get());
System.exit(0);
},
error -> {
logger.info("error count：" + count.get());
System.exit(0);
}
);
}
public static void main(String[] args) throws InterruptedException {
ReactorTest test = new ReactorTest();
test.vertxTest();
// test.webFluxTest();
Thread.sleep(10000);
}
}
值得注意的一点是，当我们使用线程池时会设置线程池大小，但是使用这两个框架时，一般是不用显式设置线程池的。
Vert.x 示例中的 setMaxPoolSize 参数是指设置HTTP连接池大小， Spring WebFlux 中的 maxConnections 参数也是指设置HTTP连接池大小。
至于线程池，一般都是默认 CPU 核心 * 2，因为我们的性能瓶颈在 IO，所以使用很少的线程也完全足以应付大量的请求（Node JS 还是单线程模型，也不妨碍它做后端服务器），也这是 Netty 单机十万并发的由来。
接下来大家可以猜一下，执行结果是什么？
Vert.x 的 count 结果是 1000。
Spring WebFlux 的结果是 256。
理论上来说经过我们上面的介绍，这里 Spring WebFlux 的结果也应该是 1000，留个思考题，为什么会这样？
总之，大家可以看到，用默认的 CPU * 2 的线程数，Vert.x 却同时执行了 1000 个任务，这在传统线程池的方案中是完全不可能的。
这，就是响应式编程。
几点补充
在张哈希张哥（掘金张哥真多～）的提醒下，我觉得有必要做几点补充。
- 宽带和网卡也是制约你能同时发起多少请求的限制，并不只是代码。
- 能够同时发起的请求数是根据连接池的配置来的，但是第三方可能也对你有限制，对方可能限制你的链接数量。
- 如果你的连接数量过大，CDN 也可能会拦截你的连接。
同时也感谢张哥和他的群友们在这个问题上对我的帮助，张哥也发文章简单记录了一下，点击这里。
感谢大家能看到这，同时也希望大家能对本篇点点赞，点赞过 100 一周内更新更多高级 Java 知识，有任何问题都可以在评论区一块讨论，祝有好收获。
注：本文中的唯一图片来自 Vert.x China。