介绍

反应式编程代表了我们对应用程序执行模型的看法的改变。在响应式应用程序中，执行不遵循一个请求由一个线程处理的线性模型，而是以事件驱动和非阻塞的方式处理多个请求。

反应式编程提供了一种基于事件的异步流式编程模型，可以处理来自单个或多个客户端的大量并发请求。反应式应用程序通常需要少量线程来垂直扩展，而不是水平扩展。使用响应式编程设计和实现应用程序使应用程序能够最大限度地利用 CPU，从而使应用程序比传统的 Java Web 应用程序具有更高的性能和效率。

反应式编程的特点：

1.异步

2.非阻塞

3. 事件驱动的数据处理方法（事件驱动架构）

术语“反应式”是指围绕对变化做出反应而构建的编程模型，例如对 I/O 事件做出反应的网络组件、对鼠标事件做出反应的 UI 控制器等。从这个意义上说，非阻塞是反应性的，因为我们现在不是被阻塞，而是在操作完成或数据可用时对通知做出反应。（将看到如何发生这种情况的示例）

反应式编程的实现

1. 反应式流（Reactive Streams）

2. RxJava 2.0（最初由Netflix开发，后来开源）

3. Project Reactor（本文）

4.Vert.x

5.其他

Project Reactor的特点：

1. I/O 操作的异步和非阻塞

2. 背压

3.错误处理

4. 处理来自单个或多个客户端的大量并发请求。

5.函数式编程

6. 反应式流是基于推送的，从而提高了性能

7.简单易用的 API （用于 [0|1] 元素） 和（用于 [N] 元素）

I/O 调用非阻塞

反应式和非阻塞式的主要预期好处是能够使用少量固定数量的线程和更少的内存进行扩展。

阻塞可能是浪费，总的来说，有两种方法可以提高程序的性能：

1.并行化以使用更多线程和更多硬件资源。

2.在如何使用当前资源方面寻求更高的效率。

如果仔细观察，一旦程序涉及一些延迟（特别是 I/O，例如数据库请求或网络调用），就会浪费资源，因为线程（可能很多线程）现在处于空闲状态，等待数据。

阻塞请求处理(每个连接一个线程)

非阻塞异步请求处理(事件回调)

事件处理是通过使用：Channel、Buffer 和 Selector 来实现的。

下面列出了一些常用的Channel：

• FileChannel → 通过文件读写
• SocketChannel → 通过 TCP 套接字读写
• ServerSocketChannel →监听客户端的 TCP 连接，为每个 TCP 连接创建新的 SocketChannel
• DatagramChannel →通过UDP协议读写

Buffer可以看成是一个数据容器，可以用数组来实现。无论是从通道读取还是写入通道，都必须先将数据放入缓冲区。

线程选择器通道

Selector是 Java NIO 中的核心类，它监视和处理在多个注册的 Channel 中发生的感兴趣的 IO 事件。通过这种机制，我们可以只用一个线程来维护多个连接。只有当这些连接之间真正发生了 IO 事件时，才会真正调用 IO 流程逻辑。每次有新连接进来时，不需要启动一个新线程，因此可以显着降低系统负载。

与 Selector 一起，SelectionKey 是另一个重要的类，它代表一个到达的事件。这两个类构成了 NIO 服务器的关键逻辑。

上图显示了在单个线程上运行的 Selector 对多个通道（或连接）的监控。

背压

背压就是下游比上游处理的慢时的一种反馈信号。

Reactive Streams 中的背压概念既优雅又强大。它将允许在响应式应用程序中使用缓慢的消费者，而不会“阻塞”太多信息。可以缓冲未处理的数据。

当消费者在生产者上订阅自己时，它将获得一个订阅。这将启用从数据流的消费者到其生产者的反馈机制。通过它，消费者可以指示他能够处理多少数据事件。

当 Consumer 发出可以处理 5 个数据事件的信号时，Producer 将使用 onNext 方法最多调用 Consumer 5 次。消费完这 5 个事件后，Consumer 会向 Producer 请求额外的事件，直到发生 onComplete 或 onError 调用。

简单易用的 API - Mono（用于 [0|1] 元素）和 Flux（用于 [N] 元素）

Spring WebFlux 围绕 2 个 API 的 Flux 和 Mono 展开 ……

Flux，一个 0-N 项的异步序列

下图显示了如何Flux转换过程：

Mono，异步 0-1 结果

下图显示了如何Mono转换过程：

代码示例：

非反应性代码

@GetMapping("/tweets-blocking")
public List<Tweet> getTweetsBlocking() {
  log.info("Starting BLOCKING Controller!”);
  final String uri = getSlowServiceUri();
  ResponseEntity<List<Tweet>> response = new RestTemplate().exchange(uri, HttpMethod.GET, null, new ParameterizedTypeReference<List<Tweet>>(){});
  List<Tweet> result = response.getBody();
  result.forEach(tweet -> log.info(tweet.toString()));
  log.info("Exiting BLOCKING Controller!");
  return result;
}

反应式代码

@GetMapping(value = "/tweets-non-blocking")
public Flux<Tweet> getTweetsNonBlocking() {
    log.info("Starting NON-BLOCKING Controller!");
    WebClient.create()
        .get()
        .uri(getSlowServiceUri())
        .retrieve()
        .bodyToFlux(Tweet.class)
        .subscribe(tweet -> log.info(tweet.toString()));
    log.info("Exiting NON-BLOCKING Controller!");
    return tweetFlux;
}

两者不同的一点是：反应式中直到订阅的时候（subscribes）才会处理，即延迟执行/懒式执行。

总结

Spring Boot 反应式（Spring Webflux）与非反应式堆栈的比较

有什么优点和缺点？

优点

• 代码更简洁，更简洁
• 更容易阅读（一旦你掌握了它）
• 更容易扩展（管道任何操作）
• 更好的错误处理
• 事件驱动的启发 -> 与流（Kafka、RabbitMQ 等）配合得很好
• 背压（客户端可以控制流量）
• 非阻塞 I/O 操作（这意味着更好的响应时间）

缺点

· 大部分时间存储数据流需要更多内存（因为它是基于一段时间的流）

• 一开始可能会觉得学习非常规（需要一切都是流）

· 与传统java风格相比不同的编程风格

· 一切都必须围绕 Mono/Flux

什么时候使用？

实现高吞吐量

系统中的大量数据流

什么时候不使用

当系统内没有太多数据需要处理时

什么是反应式系统？

反应式系统与反应式编程不同。反应式编程用于代码级别，而反应式系统处理架构。

·响应能力：对用户可用，并且无论发生什么（过载、故障等），都准备好响应他们。

·弹性：保持不受故障、中断和极高负载的影响。

·弹性：有效利用资源并平衡机器性能——垂直扩展或缩减——或轻松调节涉及的机器数量——水平扩展或缩减——取决于负载。

·消息驱动特性：通过向可寻址的接收者发送不可变消息来实现完全无阻塞的通信。

官网：https://projectreactor.io/docs/core/release/reference/#getting-started-introducing-reactor