前言

前面的章节我们讲了REST。本节，继续微服务专题的内容分享，共计16小节，分别是：

• 微服务专题01-Spring Application
• 微服务专题02-Spring Web MVC 视图技术
• 微服务专题03-REST
• 微服务专题04-Spring WebFlux 原理
• 微服务专题05-Spring WebFlux 运用
• 微服务专题06-云原生应用（Cloud Native Applications）
• 微服务专题07-Spring Cloud 配置管理
• 微服务专题08-Spring Cloud 服务发现
• 微服务专题09-Spring Cloud 负载均衡
• 微服务专题10-Spring Cloud 服务熔断
• 微服务专题11-Spring Cloud 服务调用
• 微服务专题12-Spring Cloud Gateway
• 微服务专题13-Spring Cloud Stream (上)
• 微服务专题14-Spring Cloud Bus
• 微服务专题15-Spring Cloud Stream 实现
• 微服务专题16-Spring Cloud 整体回顾

本节内容重点为：

• Reactive 原理：理解 Reactive 本质原理，解开其中的奥秘

• WebFlux 使用场景：介绍 WebFlux 与 Spring Web MVC 的差异，WebFlux 真实的使用场景

• WebFlux 整体架构：介绍 WebFlux、Netty 与 Reactor 之间的关系，对于 Spring Web MVC 架构深入理解

Reactive 原理

关于Spring WebFlux的基本情况这里不再赘述，与Spring MVC不同，它不需要Servlet API，完全异步和非阻塞， 并通过Reactor项目实现Reactive Streams规范。 并且可以在诸如Netty，Undertow和Servlet 3.1+容器的服务器上运行。

关于 Reactive 的一些讲法

其中笔者挑选了以下三种出镜率最高的讲法：

Q：Reactive 是异步非阻塞编程（错误）
A：Reactive 是同步/异步非阻塞编程

Q： Reactive 能够提升程序性能
A：大多数情况是没有的，少数可能能会，参考测试用例地址：https://blog.ippon.tech/spring-5-webflux-performance-tests/

Q： Reactive 解决传统编程模型遇到的困境
A： 也是错的，传统困境不需，也不能被 Reactive

传统编程模型中的某些困境

Reactor 认为阻塞可能是浪费的

http://projectreactor.io/docs/core/release/reference/#_blocking_can_be_wasteful
将以上 Reactor 观点归纳如下，它认为：

• 阻塞导致性能瓶颈和浪费资源

1. 任何代码都是阻塞（指令是串行）
2. 非阻塞从实现来说，就是回调

   当前不阻塞，事后来执行

• 增加线程可能会引起资源竞争和并发问题

  通用问题

• 并行的方式不是银弹（不能解决所有问题）

来一个Spring-Event的 DEMO 的感受一下：

   public static void main(String[] args) {// 默认是同步非阻塞SimpleApplicationEventMulticaster multicaster = new SimpleApplicationEventMulticaster();// 构建线程池ExecutorService executor = Executors.newSingleThreadExecutor();// 切换成异步非阻塞multicaster.setTaskExecutor(executor);// 增加事件监听器multicaster.addApplicationListener(event -> { // Lambda 表达// 事件监听System.out.printf("[线程 : %s] event : %s\n",Thread.currentThread().getName(), // 当前执行线程名称event);});// 广播事件multicaster.multicastEvent(new PayloadApplicationEvent("Hello,World", "Hello,World"));// 关闭线程池executor.shutdown();}

Reactor 认为异步不一定能够救赎

再次将以上观点归纳，它认为：

• Callbacks 是解决非阻塞的方案，然而他们之间很难组合，并且快速地将代码引导至 “Callback Hell” 的不归路
• Futures 相对于 Callbacks 好一点，不过还是无法组合，不过 CompletableFuture 能够提升这方面的不足

串行与并行的效率测试

demo1、串行测试：

public class DataLoader {public final void load() {long startTime = System.currentTimeMillis(); // 开始时间doLoad(); // 具体执行long costTime = System.currentTimeMillis() - startTime; // 消耗时间System.out.println("load() 总耗时：" + costTime + " 毫秒");}protected void doLoad() { // 串行计算loadConfigurations();    //  耗时 1sloadUsers();                  //  耗时 2sloadOrders();                // 耗时 3s} // 总耗时 1s + 2s  + 3s  = 6sprotected final void loadConfigurations() {loadMock("loadConfigurations()", 1);}protected final void loadUsers() {loadMock("loadUsers()", 2);}protected final void loadOrders() {loadMock("loadOrders()", 3);}private void loadMock(String source, int seconds) {try {long startTime = System.currentTimeMillis();long milliseconds = TimeUnit.SECONDS.toMillis(seconds);Thread.sleep(milliseconds);long costTime = System.currentTimeMillis() - startTime;System.out.printf("[线程 : %s] %s 耗时 :  %d 毫秒\n",Thread.currentThread().getName(), source, costTime);} catch (InterruptedException e) {throw new RuntimeException(e);}}public static void main(String[] args) {new DataLoader().load();}}

demo1运行结果：

demo2、并行测试：

public class ParallelDataLoader extends DataLoader {protected void doLoad() {  // 并行计算ExecutorService executorService = Executors.newFixedThreadPool(3); // 创建线程池CompletionService completionService = new ExecutorCompletionService(executorService);completionService.submit(super::loadConfigurations, null);      //  耗时 >= 1scompletionService.submit(super::loadUsers, null);               //  耗时 >= 2scompletionService.submit(super::loadOrders, null);              //  耗时 >= 3sint count = 0;while (count < 3) { // 等待三个任务完成if (completionService.poll() != null) {count++;}}executorService.shutdown();}  // 总耗时 max(1s, 2s, 3s)  >= 3spublic static void main(String[] args) {new ParallelDataLoader().load();}}

demo2运行结果：

以上测试流程图：

关于串行与并行的理解：

并行+join：我们知道，在并发编程里，join()方法可以等待线程销毁，说白了，可以上多线程顺序执行，常见的CountDownLatch则是通过AQS -> (状态位、队列 Integer）效果实现顺序执行。

线程测试：

  public static void main(String[] args) throws InterruptedException {println("Hello,World 1");AtomicBoolean done = new AtomicBoolean(false);final  boolean isDone;// volatile 易变，线程安全（可见性）// final 不变，线程安全（一直不变）// final + volatile  = impossibleThread thread = new Thread(() -> {// 线程任务println("Hello,World 2020");// CASdone.set(true);  // 不通用});thread.setName("sub-thread");// 线程名字thread.start(); // 启动线程// 线程 join() 方法thread.join(); // 等待线程销毁println("Hello,World 2");}private static void println(String message) {System.out.printf("[线程 : %s] %s\n",Thread.currentThread().getName(), // 当前线程名称message);}

运行结果：

关于Java 8 Lambda 表达式里使用boolean类型变量问题：通过我们在一个事件里使用标记位采用boolean，但是如果这个事件被lambda所封装，就不能简单的使用boolean，即使使用final修饰，所以在上面的demo里采用AtomicBoolean（线程安全）作为标记位。

至于说final volatile同时修饰为什么不行？原因很简单，因为两个关键字本身就是对立的：

volatile 易变，线程安全（可见性）
final 不变，线程安全（一直不变）

CompletableFuture

Future 的局限性

• get() 方法是阻塞的
  
• Future 没有办法组合
  • 任务Future之间有依赖关系

  通俗讲就是第一步的结果，是第二部的输入

CompletableFuture 的功能列举

• 提供异步操作
• 提供 Future 链式操作
• 提供函数式编程

CompletableFuture测试

public class CompletableFutureDemo {public static void main(String[] args) {println("当前线程");// Reactive programming// Fluent 流畅的// Streams 流式的CompletableFuture.supplyAsync(() -> {println("第一步返回 \"Hello\"");return "Hello";}).thenApplyAsync(result -> { // 异步？println("第二步在第一步结果 +\",World\"");return result + ",World";}).thenAccept(CompletableFutureDemo::println) // 控制输出.whenComplete((v, error) -> { // 返回值 void, 异常 -> 结束状态println("执行结束!");}).join() // 等待执行结束;}private static void println(String message) {System.out.printf("[线程 : %s] %s\n",Thread.currentThread().getName(), // 当前线程名称message);}
}

上述代码采用的是命令编程方式（Imperative programming）
命令编程方式最大的特点是流程编排，其优势在于：

• 大多数业务逻辑是数据操作
• 消费类型 Consumer
• 转换类型 Function
• 提升/减少维度 map/reduce/flatMap

传统的编程模式：三段式编程

        try {// 1、业务执行// action} catch (Exception e) {// 2、异常处理// error} finally {// 3、执行完成//  complete}

CompletableFutureDemo 运行结果为：

以上流程图为：

函数式编程 + Reactive

• Reactive programming
• 编程风格
  • Fluent 流畅的
  • Streams 流式的
• 业务效果
  • 流程编排
  • 大多数业务逻辑是数据操作
• 函数式语言特性（Java 8+）
  • 消费类型 Consumer
  • 生产类型 Supplier
  • 转换类型 Function
  • 判断类型 Predicate
  • 提升/减少维度 map/reduce/flatMap

来一个steam流式处理操作demo：

        Stream.of(0, 1, 2, 3, 4, 5, 6, 7, 8, 9) // 0-9 集合.filter(v -> v % 2 == 1) // 判断数值->获取奇数.map(v -> v - 1) // 奇数变偶数.reduce(Integer::sum) // 聚合操作.ifPresent(System.out::println) // 输出 0 + 2 + 4 + 6 + 8

以上操作是不是非常直观? 就是一次性的将数据处理完成！

其实不论Java/C#/JS/Python/Scale/Koltin语言，都在使用这种操作(Reactive/Stream模式)

Stream 是迭代器（ Iterator ）模式，数据已完全准备，拉模式（Pull）
Reactive 是观察者（Observer）模式，来一个算一个，推模式（Push），当有数据变化的时候，作出反应（Reactor）

Reactive Programming

Reactive Programming 作为观察者模式的延伸，不同于传统的命令编程方式同步拉取数据的方式，如迭代器模式。而是采用数据发布者同步或异步地推送到数据流（Data Streams）的方案。当该数据流（Data Streams）定于这坚挺到传播变化时，立即做出响应动作。在实现层面上，Reactive Programming 可结合函数式编程简化面向对象语言语法的臃肿性，屏蔽并发实现的复杂细节，提供数据流的有序操作，从而达到提升代码的可读性，以及减少Bugs出现的目的。同时，Reactive Programming结合背压（Backpressure）的技术解决发布端生成数据的速率高于订阅端消费的问题。

WebFlux 使用场景

先上一个demo：

  public static void main(String[] args) throws InterruptedException {Flux.just(0, 1, 2, 3, 4, 5, 6, 7, 8, 9) // 直接执行.filter(v -> v % 2 == 1) // 判断数值->获取奇数.map(v -> v - 1) // 奇数变偶数.reduce(Integer::sum) // 聚合操作.subscribeOn(Schedulers.elastic())
//                .subscribeOn(Schedulers.parallel())
//                .block());.subscribe(ReactorDemo::println) // 订阅才执行;Thread.sleep(1000);}private static void println(Object message) {System.out.printf("[线程 : %s] %s\n",Thread.currentThread().getName(), // 当前线程名称message);}

执行结果：

我们发现WebFlux 的特性：

• 长期异步执行，一旦提交，慢慢操作。

那么是否适合 RPC 操作？

• 任务型的，少量线程，多个任务长时间运作，达到伸缩性。

Flux 和 Mono 是 Reactor 中的两个基本概念。
Mono：单数据 Optional 0:1, RxJava : Single
Flux : 多数据集合，Collection 0:N , RxJava : Observable

同样，再举一个栗子，看看WebFlux 与SpringMVC的性能比较：

@RestController
public class WebFluxController {@RequestMapping("")public Mono<String> index() {// 执行计算println("执行计算");Mono<String> result =  Mono.fromSupplier(() -> {println("返回结果");return "Hello,World";});return result;}private static void println(String message) {System.out.printf("[线程 : %s] %s\n",Thread.currentThread().getName(), // 当前线程名称message);}
}

测试结果（这里使用WebFluxApplication作为启动类，并在浏览器访问http://localhost:8080/）：

相对SpringMVC来说，WebFlux执行效率不见得比SpringMVC要快，只是WebFlux在多线程异步处理方面比较友好，使得其具有更好伸缩性，其底层还是基于并发编程。

使用场景

1. 函数式编程

2. 非阻塞（同步/异步）

3. 远离 Servlet API
   Servlet
HttpServletRequest

4. 不再强烈依赖 Servlet 容器（兼容）
   Tomcat
Jetty

实际上很多技术基于Reactor做了实现，Reactor真的可以引领下一代么？
Spring Cloud Gateway -> Reactor

Spring WebFlux -> Reactor

Zuul2 -> Netty Reactive

WebFlux 整体架构

后记

本节示例代码：https://github.com/harrypottry/microservices-project/tree/master/spring-reactive

更多架构知识，欢迎关注本套Java系列文章：Java架构师成长之路