背景

随着业务的发展，微服务应用的流量越来越大，使用到的资源也越来越多。

在微服务架构下，大量的应用都是 SpringCloud 分布式架构，这种架构，总体是全链路同步模式。

同步编程模式不仅造成了资源的极大浪费，并且在流量发生激增波动的时候，受制于系统资源而无法快速的扩容。

全球后疫情时代，降本增效是大背景。

如何降本增效？

可以通过技术升级，全链路同步模式 ，升级为 全链路异步模式。

尼恩作为40岁资深老架构师，带大家来做一把全链路异步模式改造，给大家看看研究成果，一定会惊到大家目瞪口呆。

全链路同步模式架构图

先回顾一下全链路同步模式架构图

全链路同步模式 ，如何升级为 全链路异步模式， 就是一个一个 环节的异步化。

40岁老架构师尼恩，持续深化自己的3高架构知识宇宙，当然首先要去完成一次牛逼的全链路异步模式 微服务实操，下面是尼恩的实操过程、效果、压测数据。

全链路异步模式

网关纯异步化（提升 9倍以上）

网关层的特点：

• 不需要访问业务数据库只做协议转换和流量转发
• 特点是 IO 密集型，特别适合纯异步的架构，可以极大的节省资源。

如何进行网关异步化？

使用高性能的通信框架Netty，这是一个基于NIO 非阻塞IO+ Reactor 纯异步线程模型的纯异步化框架。

网关的技术选型主要有 zuul，SpringCloud GetWay 。

• zuul 1虽然使用的同步io，zuul2它也是使用异步的netty，但是没有和SpringCloud 框架集成
• springcloud getway 它是基于spring 5.0 、spring boot 2.0 和spring reacter，为微服务提供一个简单有效的网关API路由接口。和SpringCloud 框架完美集成，目标是为了代替zuul

SpringCloud GetWay 是基于webFlux框架实现的，而WebFlux框架底层则使用了高性能的Reactor模式通信框架Netty。所以最终还是基于IO的王者组件Netty。

如果大家使用Zuul 1，那么 升级为 SpringCloud GetWay，性能可以提升 9倍以上，

以上结论，是来自于尼恩的读者群（50+），如有疑问，可以来单挑。

总体来说，这个环节，是纯异步化最容易的。

这个环节，大部分已经升级到了 springcloud getway 已经使用了纯异步的架构；

Web 服务异步化（2W并发场景提升 20倍以上）

Web 服务作为微服务体系内的重要组成，服务节点众多，

Springboot的Web 服务默认为 Tomcat + Servlet 不支持纯异步化编程，

Tomcat + Servlet模式的问题：总体上没有使用Reactor 反应器模式， 每一个请求是阻塞处理的，属于同步 Web 服务类型。

Servlet 有异步的版本，可惜没有用起来。具体请参考 40岁老架构师尼恩为大家整理的深度文章：

京东一面：20种异步，你知道几种？含协程

所以：跑在 大家生产环境上的，还是Tomcat + Servlet 同步 Web 服务。

如何实现 Web 服务异步化：

• 方式一：基于Netty 实现web服务
• 方式二：使用 WebFlux （还是 Netty 实现web服务）

Spring WebFlux是一个响应式堆栈 Web 框架 ，它是完全非阻塞的，支持响应式流(Reactive Stream)背压，并在Netty，Undertow和Servlet 3.1 +容器等服务器上运行

我们再来看一下对于 WebFlux 的对比测试数据 （来自于参考文献1）：

可见，非阻塞的处理方式规避了线程排队等待的情况，从而可以用少量而固定的线程处理应对大量请求的处理。

还有更绝的，小伙伴又一步到位直接测试了一下20000用户的情况：

1. 对 mvc 的测试由于出现了许多的请求fail，最终以失败告终；
2. 而 WebFlux 应对20000用户已然面不改色心不慌，吞吐量达到7228 req/sec

注意：正好是10000用户下的两倍，绝对是真实数据！也就是说, 2W并发场景提升 20倍以上

95%响应时长仅117ms。

最后，再给出两个吞吐量和响应时长的图，更加直观地感受异步非阻塞的WebFlux是如何一骑绝尘的吧：

此时，我们更加理解了Nodejs的骄傲，不过我们大Java语言也有了Vert.x和现在的Spring WebFlux。

RPC 调用异步化（提升 9倍以上）

异步RPC 调用，等待upstream 上游 response 返回时，线程不处于block 状态

作为微服务架构中数据流量最大的一部分，RPC 调用异步化的收益巨大；

RPC 调用主要的框架有：

特点是：

• feign 是同步IO 、阻塞模式的同步 RPC框架
• dubbo 是基于Netty的非阻塞IO + Reactor 反应堆线程模型的 异步RPC框架

40岁老架构师尼恩，完成了 SpringCloud + Dubbo RPC 的集成，在同一个微服务下，同时使用了Feign + Dubbo

然后进行了性能的对比验证

dubbo 的压测数据

wrk -t8 -c200 -d30s --latency  http://cdh1:18081/dubbo-consumer-demo/user/detail/v1?userId=1

[root@centos1 src]# wrk -t8 -c200 -d30s --latency  http://cdh1:18081/dubbo-consumer-demo/user/detail/v1?userId=1
Running 30s test @ http://cdh1:18081/dubbo-consumer-demo/user/detail/v1?userId=1
  8 threads and 200 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency    30.10ms   45.68ms 644.45ms   95.43%
    Req/Sec     1.12k   465.63     2.36k    66.87%
  Latency Distribution
     50%   18.94ms
     75%   28.43ms
     90%   46.21ms
     99%  283.56ms
  264316 requests in 30.07s, 148.47MB read
Requests/sec:   8788.96
Transfer/sec:      4.94MB

feign 的压测数据

wrk -t8 -c200 -d30s --latency  http://cdh1:18081/dubbo-consumer-demo/echo/variable/11

[root@centos1 src]# wrk -t8 -c200 -d30s --latency  http://cdh1:18081/dubbo-consumer-demo/echo/variable/11
Running 30s test @ http://cdh1:18081/dubbo-consumer-demo/echo/variable/11
  8 threads and 200 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency   321.50ms  294.59ms   2.00s    61.77%
    Req/Sec    87.18     43.39   232.00     67.00%
  Latency Distribution
     50%  309.06ms
     75%  503.06ms
     90%  687.99ms
     99%    1.21s
  20495 requests in 30.10s, 7.64MB read
  Socket errors: connect 0, read 0, write 0, timeout 49
Requests/sec:    680.90
Transfer/sec:    259.99KB

从数据来看， dubbo rpc 是feign rpc 性能10倍

当然，感兴趣的小伙伴，也可以自己实操一下，更有感触。

Cache异步化（提升2倍+）

Cache Aside 缓存模式，是大家通用的Cache使用方式，Cache纯异步的架构，必须使用异步存储层客户端，

主要有：

• Redisson
• Lettuce

Redisson、Lettuce如何选型？请参考40岁老架构师尼恩的文章：

Jedis那么低性能，还在用？赶紧换上 lettuce 吧

40岁老架构师尼恩，完成了自己的开发脚手架Crazy-SpringCloud 的Cache异步化，经过对比验证，性能提升足足2倍多

使用Lettuce的场景：

[root@centos1 ~]# wrk -t8 -c200 -d30s --latency   http://192.168.56.121:7703/uaa-react-provider/api/userCacheAside/detail/v1?userId=1
Running 30s test @ http://192.168.56.121:7703/uaa-react-provider/api/userCacheAside/detail/v1?userId=1
  8 threads and 200 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency    18.29ms   13.56ms 213.57ms   89.56%
    Req/Sec     1.51k   504.74     4.26k    72.86%
  Latency Distribution
     50%   14.56ms
     75%   19.92ms
     90%   31.20ms
     99%   76.70ms
  359546 requests in 30.10s, 53.15MB read
Requests/sec:  11945.39
Transfer/sec:      1.77MB

使用jedis的场景

 wrk -t8 -c200 -d30s --latency  http://192.168.56.121:7702/uaa-provider/api/user/detailCacheAside/v1?userId=1
  
 [root@centos1 src]#  wrk -t8 -c200 -d30s --latency  http://192.168.56.121:7702/uaa-provider/api/user/detailCacheAside/v1?userId=1
Running 30s test @ http://192.168.56.121:7702/uaa-provider/api/user/detailCacheAside/v1?userId=1
  8 threads and 200 connections
  Thread Stats   Avg      Stdev     Max   +/- Stdev
    Latency    42.20ms   44.79ms   1.11s    93.41%
    Req/Sec   683.30    245.08     1.85k    67.39%
  Latency Distribution
     50%   32.65ms
     75%   48.30ms
     90%   72.32ms
     99%  199.81ms
  162271 requests in 30.09s, 71.96MB read
Requests/sec:   5393.75
Transfer/sec:      2.39MB

吞吐量 从 5000 提升到 10000

99% 响应时间 从199.81ms降低到 76.70ms

DB的异步化 （假装提升10倍）

数据操作是每个请求调用链的 终点，纯异步的架构必须使用异步存储层客户端，

比如说，可以使用纯异步化的框架 Spring Data R2DBC

在尼恩的Crazy-SpringCloud 脚手架 纯异步化 改造中，没有对 的DB 进行异步化改造，为啥呢？DB是一个低吞吐的物种，对于DB而已，请求太多，反而忙不过来，造成整体的性能下降。

所以，尼恩没有对DB进行纯异步化改造，反而是进行隔离和保护：

• 参考 Hystrix舱壁模式， 通过 DB 的操作进行 线程池隔离，
• 使用 手写 Hystrix Command 的方式，进行 DB 操作的 高压防护。

控制线程数和请求数，保护不至于拖垮DB

由于高压防护，在高并发场景能快速失败，所以肯定提升不止10倍，不过是假装提升10倍

纯异步与伪异步

异步调用目的在于防止当前业务线程被阻塞。

伪异步将任务包装为Runnable 放入另一个线程执行并等待，当前Biz 线程不阻塞；

纯异步为响应式编程模型，通过IO 实践驱动任务完成。

两个概念很重要，这里不做赘述，具体请参考 40岁老架构师尼恩为大家整理的深度文章：

京东一面：20种异步，你知道几种？含协程

全链路异步，让你的性能优化10倍+

降本增效时代，大家行动起来吧，对SpringCloud 微服务进行一场性能提升革命

想尽办法，让一台服务器，发挥10台的价值

特别提示：

在尼恩的全链路异步 改造的过程中， 大量使用 了 响应式编程。关于响应式编程的知识，请参考尼恩的 深度文章：

Flux、Mono、Reactor 实战（史上最全）：https://blog.csdn.net/crazymakercircle/article/details/124120506

遗憾的是，响应式编程非常复杂，下面带你学习Spring响应式编程

一文带你搞懂Spring响应式编程

了解响应式编程，首先我们需要了解函数式操作和Stream的操作，下面我们简单的复习一下喽。

1.1 常用函数式编程

函数式接口中

我们先来回顾一下Java中的函数式接口。常见的有以下几种

Consumer 一个输入，无输出
Supplier 无输入，有输出
Function 输入T，输出R
BiFunction 输入T,U 输出R
Predicate 有输入，输出boolean类型

上面的简单函数式接口示例如下：

Consumer consumer = (i)-> System.out.println("this is " + i);
consumer.accept("consumer");
Supplier supplier = () -> "this is supplier";
System.out.println(supplier.get());
Function function = (i) -> i*i;
System.out.println(function.apply(8));
BiFunction biFunction = (i,j)-> i+"*"+j+"="+i*j;
System.out.println(biFunction.apply(8,8));
Predicate predicate = (i) -> i.intValue()>3;
System.out.println(predicate.test(5));

其执行结果如下：

this is consumer
this is supplier
64
8*8=64
true

1.2 Stream操作

对Stream进行操作，主要有几个关键点：

• 生成流
• 流的中间操作其中中间操作可以有多个，中间操作会返回一个新的流（如 map ,filter,sorted等），然后交给下一个流方法使用。
• 流的终结操作终结操作只有一个。终结操作执行后，流就到了终止状态，无法被操作 （如forEach，toArray , findFirst 等）。

创建流的示例：

String[] strArray = {"ss","ss","","sdffg"};
Arrays.stream(strArray).forEach(System.out::println);
Arrays.asList(strArray).stream().forEach(System.out::println);
Stream.of(strArray).forEach(System.out::println);
Stream.iterate(1,(i) -> i+1).limit(10).forEach(System.out::println);
Stream.generate(() -> new Random().nextInt(10)).limit(10).forEach(System.out::println);

简单的流处理示例：

String[] strArray1 = {"ss","ss","","sdffg","bca-de","fff"};
String collect = Stream.of(strArray1)
.filter(i -> !i.isEmpty())//过滤空字符串
.sorted() //排序
.limit(1) //只取第一个元素
.map(i -> i.replace("-", ""))//替换 "-"
.flatMap(i -> Stream.of(i.split("")))//将字符拆成字符数组
.sorted() //排序
.collect(Collectors.joining());//将字符拼接组合到一起
System.out.println(collect);//最后输出abcde

2. Java响应式编程

响应式编程会用到一个发布者和一个订阅者，然后通过订阅关系完成数据流的传输。订阅关系中可以处理一些背压问题，即调节消费者与生产者之间的供需平衡，让整个程序达到最大效率。

Java9中java.util.concurrent.Flow接口提供响应式流编程类似的功能。

下面我们实现一个基于Java 响应式编程的示例：

其中有三个简单步骤：

• 建立生产者
• 构建消费者
• 消费者订阅生产者
• 生产者生产内容

SubmissionPublisher publisher = new SubmissionPublisher<>();//建立生产者
Flow.Subscriber subscriber = new Flow.Subscriber() {...};//建立消费者 (其中的实现放到下面)
publisher.subscribe(subscriber);//订阅关系
for (int i = 0; i < 10; i++) {
publisher.submit("test reactive java : " +i); //生产者生产内容
}

消费者全部代码如下：

Flow.Subscriber subscriber = new Flow.Subscriber() {
Flow.Subscription subscription;
@Override
public void onSubscribe(Flow.Subscription subscription) {
System.out.println("Subscription establish first ");
this.subscription = subscription;
this.subscription.request(1);
}
@Override
public void onNext(Object item) {
subscription.request(10);
System.out.println("receive : "+ item);
}
@Override
public void onError(Throwable throwable) {
System.out.println(" onError ");
}
@Override
public void onComplete() {
System.out.println(" onComplete ");
}
};

其中onSubscribe方法表示建立订阅关系

onNext接受数据，并请求生产者的数据。

onError，onComplete则是error或者完成之后的处理方法。

带有中间处理器的响应式流

Reactive Stream 通常会基于如下的模型：

下面我们实现一个带有中间处理功能的响应式模型：

下面的Processor 既有发布者，又有订阅者：

public class ReactiveProcessor extends SubmissionPublisher implements Flow.Subscriber {
private Flow.Subscription subscription;
@Override
public void onSubscribe(Flow.Subscription subscription) {
System.out.println( Thread.currentThread().getName() + " Reactive processor establish connection ");
this.subscription = subscription;
this.subscription.request(1);
}
@Override
public void onNext(Object item) {
System.out.println(Thread.currentThread().getName() + " Reactive processor receive data: "+ item);
this.submit(item.toString().toUpperCase());
this.subscription.request(1);
}
@Override
public void onError(Throwable throwable) {
System.out.println("Reactive processor error ");
throwable.printStackTrace();
this.subscription.cancel();
}
@Override
public void onComplete() {
System.out.println(Thread.currentThread().getName() + " Reactive processor receive data complete ");
}
}

如上中间处理器订阅发布者， 同时消费者再订阅中间处理器。中间处理器也可以调节发布订阅的生产消费速率。

SubmissionPublisher publisher = new SubmissionPublisher<>(); //创建生产者
ReactiveProcessor reactiveProcessor = new ReactiveProcessor(); // 创建中间处理器
publisher.subscribe(reactiveProcessor); //中间处理器订阅生产者
Flow.Subscriber subscriber = new Flow.Subscriber() {...}; //创建消费者
reactiveProcessor.subscribe(subscriber); //消费者订阅中间处理器
for (int i = 0; i < 10; i++) {
publisher.submit("test reactive java : " +i); //生产者生产数据
}

通过上述生产者-> 中间处理器->消费者， 可以将生产者生产的数据全部变成大写，然后再发送给最终的消费者。

以上式Java中的reactive 编程示例。Java会不同线程来分别处理消费者与生产者的消息处理

3. Reactor

Reactor中两个比较关键的对象式Flux和Mono， 整个Spring的响应式编程均式基于projectreactor项目。Reactor是响应式编程的依赖，主要是基于JVM构建非阻塞程序。

根据Reactor的介绍，此类响应式编程的的三方库（Reactor）主要是解决一些JVM经典异步编程中的一些缺点，并且还可以专注于一些新的特性，如下：

• 可组合性与可读性 （Composability and readability）
• 可以使用丰富的运算操作符将数据作为流进行操作
• 订阅之前，不会有任何事
• 背压特性（Backpressure ），可以理解为消费者可以向生产者发送产出率过高的信号，从而调整生产速率。或者消费者可以选择一次性拉去一捆数据进行消费。
• 于并发无关的高度抽象的高级功能

其中有这么一段解释，可以形象的说明响应式编程。

Reactive的程序可以想象成车间的流水线，reactor既是流水线上的传送带，又是处理工作站。原料从一个原始的生产者出发，最终成为产品被推总给消费者。

3.1 Flux & Mono

下面我们介绍一下Flux和Mono。

在Reactor中Flux和Mono均是Publisher，即生产者。两者也有不同。Flux对象表示0到N个异步的响应序列，而Mono只代表0个（empty）或者1个结果。

Reactor官网上介绍的Flux示意如下：

Mono示意如下：

3.2 Flux Mono创建与使用

我们也可以单独引用其依赖。

使用maven依赖

io.projectreactor
reactor-core


io.projectreactor
reactor-test
test

Mono创建

分别创建空Mono和一个包含一个String的Mono，并由消费者消费打印。

Mono.empty().subscribe(System.out::println);
Mono.just("Hello Mono Java North").subscribe(System.out::print);

Flux创建

Flux创建有如下的一些方法，

• just（通过不定参数创建）
• range（从某个整数开始，往后的整数数量）
• fromArray，fromIterable，fromStream，从名称上就可以看出来，通过数组，迭代器，Stream流创建Flux

下面式一些Java代码示例

Flux.just(1,2,3,4,5).subscribe(System.out::print);
Flux.range(1,20).subscribe(System.out::print);
Flux.fromArray(new String[]{"a1","a2","a3","a4","a5","a6"}).skip(2).subscribe(System.out::print);
Flux.fromIterable(Arrays.asList(1,2,3,4,5,6,7)).subscribe(System.out::println);
Flux.fromStream(Stream.of(Arrays.asList(1,2,3,4,5,6,7))).subscribe(System.out::print);

我们再举一个generate的例子

public static  Flux generate(Callable stateSupplier, BiFunction, S> generator)

如上代码所示，generate需要一个Callable参数，而且是supplier （即没有输入值，只有一个输出）

另一个参数是BiFunction （前面我们也介绍过，需要两个输入值，一个输出值）。BiFunction中的其中一个输入值是SynchronousSink，下面我们给出一个generate创建Flux的示例。

Flux.generate(() -> 0, //提供一个初始状态值0
              (i, sink) -> {
sink.next("3*" + i + "=" + 3 * i);//使用初始值去生产一个3的乘法
if (i > 9) sink.complete();//设置停止条件
return i + 1;//返回一个新的状态值，以便在下一次的生产中使用，除非响应序列终止
}).subscribe(System.out::println);

下面我们在看一个Flux嵌套处理示例：

需求：将字符串去空格，并去重，然后排序输出。

String str = "qa ws ed rf tg yh uj i k ol p za sx dc vf bg hn jm k loi yt ";
Flux.fromArray(str.split(" "))//通过数组创建Flux
.flatMap(i -> Flux.fromArray(i.split("")))
.distinct() // 去重
.sort() //排序
.subscribe(System.out::print);

//flatMap与Stream中的flatMap类似，接受Function作为参数，输入一个值，输出一个值，此处输出均为Publisher，

以上就是Flux和Mono的一些简单介绍，同时Ractor也支持JDK中的FlowPubliser 和FlowSubscriber与 Reactor中的publisher, subscriber的适配等.

4. WebFlux

SpringBoot 2之后支持的Reactive响应式编程。

关于Reactive技术栈和经典的Servlet技术栈对比，Spring官网的这张图比较清晰。

Spring响应式编程主要依赖于Reactor第三方库，即上面讲的Flux和Mono的库。

WebFlux主要有以下几个要点：

• 反应式栈web框架
• 完全异步非阻塞
• 运行在netty，undertow，Servlet3.1 + 容器
• 核心反应式库 Reactor
• 返回 Flux 或Mono
• 支持注解和函数编程两种编程模式

Spring WebFlux示例

下面我们给出几个SpringBoot 的响应式web示例。

可以去https://start.spring.io/ 新建webflux的项目也可以。

项目中的主要依赖就是spring-boot-starter-webflux

org.springframework.boot
spring-boot-starter-webflux

基于注解的WebFlux

以下是一个最简单的基于注解的WebFlux

@GetMapping("/hello/mono1")
public Mono mono(){
return Mono.just("Hello Mono - Java North");
}
@GetMapping("/hello/flux1")
public Flux flux(){
return Flux.just("Hello Flux","Hello Java North");
}

基于函数式编程的WebFlux

创建RouterFunction，将其注入到Spring中即可。

@Bean
public RouterFunction testRoutes1() {
return RouterFunctions.route().GET("/flux/function", new HandlerFunction() {
@Override
public Mono handle(ServerRequest request) {
return ServerResponse.ok().bodyValue("hello web flux , Hello Java North");
}
}).build();
}
//上面的方法使用函数式编程替换之后如下
@Bean
public RouterFunction testRoutes() {
return RouterFunctions.route().GET("/flux/function",
request -> ServerResponse.ok()
.bodyValue("Hello web flux , Hello Java North")).build();
}

Flux与Mono的响应式编程延迟示例

下面我们编写一段返回Mono的响应式Web服务。

@GetMapping("/hello/mono")
public Mono stringMono(){
Mono from = Mono.fromSupplier(() -> {
try {
TimeUnit.SECONDS.sleep(5);
} catch (InterruptedException e) {
throw new RuntimeException(e);
}
return "Hello, Spring Reactive date time:"+ LocalDateTime.now();
});
System.out.println( "thread : " + Thread.currentThread().getName()+ " === " + LocalDateTime.now() +" ==========Mono function complete==========");
return from;
}

使用postman请求如下， 5秒钟后返回数据。后台却在5秒中之前已经处理完整个方法。

后台打印日志：

再看一组Flux

@GetMapping(value = "/hello/flux", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public Flux flux1(){
Flux stringFlux = Flux.fromStream(IntStream.range(1,6).mapToObj(i ->{
mySleep(1);//表示睡1秒
return "java north flux" + i + "date time: " +LocalDateTime.now();
}));
System.out.println("thread : " + Thread.currentThread().getName()+ " === " + LocalDateTime.now() + " ==========Flux function complete=========");
return stringFlux;
}

此次使用谷歌浏览器请求此服务：

可以发现每隔一秒就会有一条消息被生产出来。

后台完成时间同样是在一开始就完成整个方法：

通过上述对Flux 与 Mono的例子，可以好好体会一下响应式编程。

总结

本篇回顾了函数式编程，Stream操作等，然后再举例讲了Java中的Reactive编程示例， 同时也给处理Reactor三方库的Flux于Mono的示例。

最后使用了SpringBoot WebFlux 创建简单的响应式web服务。希望能让大家更好的理解响应式编程。

到此这篇关于一文带你搞懂Spring响应式编程的文章就介绍到这了,更多相关Spring响应式编程内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！