Java Reactive Web設計與實現

注： 本文是由讀者觀看 小馬哥公開課視頻過程當中的筆記整理而成。
更多Spring Framework文章可參看筆者我的github: spring-framework-lesson 。

0. 編程模型與併發模型

Spring 5實現了一部分Reactive

Spring WebFlux： Reactive Web（non-blocking servers in general）

Spring Web MVC：傳統Servlet Web（servlet applications in general）

0.1 編程模型

編程模型：阻塞、非阻塞

• NIO：同步+非阻塞，基於事件

• 非阻塞

  • 基本上採用Callback方式
  • 當時不阻塞，後續再輸出（再回調）

0.2 併發模型

• 併發模型：

  • 同步（Sync）
  • 異步（Async）

0.3 比較

• 同步+非阻塞：線程不會改變，不會切換

[線程：main] Observable 添加觀察者！ 
[線程：main] 通知全部觀察者！ 
[線程：main] 3. 收到數據更新：Hello World 
[線程：main] 2. 收到數據更新：Hello World 
[線程：main] 1. 收到數據更新：Hello World

• 異步+非阻塞：線程會被切換

[線程：main] 啓動一個JFrame窗口！ 
[線程：AWT-EventQueue-0] 銷燬當前窗口！ 
[線程：AWT-EventQueue-0] 窗口被關閉，退出程序！

使用Jconsole查看改異步非阻塞程序

等待總數一直在增長，說明異步程序一直在等待。NIO就是無限地在處理，無限地在等待。

1. Reactive概念

Reactive Programming：響應式編程，異步非阻塞就是響應式編程(Reactive Programming)，與之相對應的是命令式編程。

Reactive並非一種新的技術，不用Reactive照樣能夠實現非阻塞（同步、異步都可，推拉模式的結合），好比利用觀察者模式實現（好比Java Swing GUI技術）。

Reactive的另一種實現方式就是消息隊列。

1.1 標準概念

1.1.1 維基百科講法

https://en.wikipedia.org/wiki...

In computing, reactive programming is a declarative programming paradigm concerned with data streams and the propagation of change. With this paradigm it is possible to express static (e.g., arrays) or dynamic (e.g., event emitters) data streams with ease, and also communicate that an inferred dependency within the associated execution model exists, which facilitates the automatic propagation of the changed data flow

關鍵點：

• 聲明式的編程範式
• 數據流
• 傳播改變
• 使用靜態數組或者動態事件

1.1.2 Reactive-Streams講法

Reactive標準：http://www.reactive-streams.org/

Reactive Streams JVM實現：https://github.com/reactive-s...

Handling streams of data—especially 「live」 data whose volume is not predetermined—requires special care in an asynchronous system. The most prominent issue is that resource consumption needs to be carefully controlled such that a fast data source does not overwhelm the stream destination. Asynchrony is needed in order to enable the parallel use of computing resources, on collaborating network hosts or multiple CPU cores within a single machine.

關鍵點：

• 處理的是數據流（活躍數據，動態數據）
• 數據容量沒法預判
• 異步系統
• 資源消費須要精細控制
• data source（數據上游），stream destination（數據下游）

In summary, Reactive Streams is a standard and specification for Stream-oriented libraries for the JVM that

• process a potentially unbounded number of elements
• in sequence,
• asynchronously passing elements between components,
• with mandatory non-blocking backpressure.

1.2 實現框架

1.2.1 ReactiveX

ReactiveX：http://reactivex.io/intro.html

ReactiveX is a library for composing asynchronous and event-based programs by using observable sequences.

It extends the observer pattern to support sequences of data and/or events and adds operators that allow you to compose sequences together declaratively while abstracting away concerns about things like low-level threading, synchronization, thread-safety, concurrent data structures, and non-blocking I/O.

• 是一個類庫
• 異步
• 基於事件

• 可觀察的數據/事件序列（sequences of data and/or events）

  • 有順序的
  • 可增加的
• 設計模式上說：繼承了觀察者模式
• 數據結構上說：支持數據序列（sequences of data），其實就是數據流（streams of data），而且須要屏蔽高併發相關（線程，同步，線程安全，併發數據結構，非阻塞I/O等）

You can think of the Observable class as a 「push」 equivalent to Iterable, which is a 「pull.」 With an Iterable, the consumer pulls values from the producer and the thread blocks until those values arrive.

• Observable觀察者模式是「push」模式

• Iterable是「pull」模式

  • 數據已經準備好了，本身直接去「拉」便可
  • Iterable：For Each語句的提高
  • Iterator：迭代器（Java 5）

1.2.2 Reactor

https://projectreactor.io/doc...

Reactor is a fully non-blocking reactive programming foundation for the JVM, with efficient demand management (in the form of managing "backpressure"). It integrates directly with the Java 8 functional APIs, notably CompletableFuture, Stream, and Duration. It offers composable asynchronous sequence APIs Flux (for [N] elements) and Mono (for [0|1] elements), extensively implementing the Reactive Streams specification.

Reactor also supports non-blocking inter-process communication with the reactor-netty project. Suited for Microservices Architecture, Reactor Netty offers backpressure-ready network engines for HTTP (including Websockets), TCP, and UDP. Reactive Encoding and Decoding are fully supported.

1.3 Spring WebFlux

https://docs.spring.io/spring...

The original web framework included in the Spring Framework, Spring Web MVC, was purpose-built for the Servlet API and Servlet containers. The reactive-stack web framework, Spring WebFlux, was added later in version 5.0. It is fully non-blocking, supports Reactive Streams back pressure, and runs on such servers as Netty, Undertow, and Servlet 3.1+ containers.

• 能夠運行在Netty，Undertow，Servlet 3.1+等容器中
• Servlet 3.1也有異步處理

1.3.1 Why WebFlux？

Part of the answer is the need for a non-blocking web stack to handle concurrency with a small number of threads and scale with fewer hardware resources. Servlet 3.1 did provide an API for non-blocking I/O. However, using it leads away from the rest of the Servlet API, where contracts are synchronous ( Filter, Servlet) or blocking ( getParameter, getPart). This was the motivation for a new common API to serve as a foundation across any non-blocking runtime. That is important because of servers (such as Netty) that are well-established in the async, non-blocking space.

The other part of the answer is functional programming. Much as the addition of annotations in Java 5 created opportunities (such as annotated REST controllers or unit tests), the addition of lambda expressions in Java 8 created opportunities for functional APIs in Java. This is a boon for non-blocking applications and continuation-style APIs (as popularized by CompletableFuture and ReactiveX) that allow declarative composition of asynchronous logic. At the programming-model level, Java 8 enabled Spring WebFlux to offer functional web endpoints alongside annotated controllers.

• 異步非阻塞

  • 異步非阻塞（non-blocking web stack）
  • 少許線程數處理併發性（handle concurrency with a small number of threads）
  • 少許硬件資源提升伸縮性（scale with fewer hardware resources.）
  • Servlet 3.1 提供一個非阻塞API
• 函數式編程（functional programming）

1.3.2 Define 「Reactive」

The term, 「reactive,」 refers to programming models that are built around reacting to change — network components reacting to I/O events, UI controllers reacting to mouse events, and others. In that sense, non-blocking is reactive, because, instead of being blocked, we are now in the mode of reacting to notifications as operations complete or data becomes available.

There is also another important mechanism that we on the Spring team associate with 「reactive」 and that is non-blocking back pressure. In synchronous, imperative code, blocking calls serve as a natural form of back pressure that forces the caller to wait. In non-blocking code, it becomes important to control the rate of events so that a fast producer does not overwhelm its destination.

Reactive Streams is a small spec (also adopted in Java 9) that defines the interaction between asynchronous components with back pressure. For example a data repository (acting as Publisher) can produce data that an HTTP server (acting as Subscriber) can then write to the response. The main purpose of Reactive Streams is to let the subscriber to control how quickly or how slowly the publisher produces data.

• 指的是圍繞對變化做出反應而構建的編程模型

  • 對I / O事件作出反應的網絡組件
  • 對鼠標事件作出反應的UI控制器等。
• 非阻塞在某種程度上說就是Reactive（In that sense, non-blocking is reactive）
• 非阻塞背壓
• 小的規範

1.3.2 Performance

Performance has many characteristics and meanings. Reactive and non-blocking generally do not make applications run faster. They can, in some cases, (for example, if using the WebClient to execute remote calls in parallel). On the whole, it requires more work to do things the non-blocking way and that can increase slightly the required processing time.

The key expected benefit of reactive and non-blocking is the ability to scale with a small, fixed number of threads and less memory. That makes applications more resilient under load, because they scale in a more predictable way. In order to observe those benefits, however, you need to have some latency (including a mix of slow and unpredictable network I/O). That is where the reactive stack begins to show its strengths, and the differences can be dramatic.

• 並非爲了變得更快

  • 通常而言，並非爲了使應用程序運行更快（Reactive and non-blocking generally do not make applications run faster）
  • 爲了非阻塞須要額外的工做可能會增長處理時間（it requires more work to do things the non-blocking way and that can increase slightly the required processing time）

• 關鍵的一個好處（The key expected benefit）

  • 利用固定、少許的線程，小的內存提供伸縮性的能力（reactive and non-blocking is the ability to scale with a small, fixed number of threads and less memory）
• 某種程度上，WebFlux並不適合Web請求

1.4 特性

• 異步
• 非阻塞
• 事件驅動
• 可能有背壓（back pressure）

• 多路複用

  • 這點是NIO中的特色，Reactive是整個編程範式的概念
• 高吞吐

1.5 目的

防止回調地獄（Callback Hell）

Callbacks are hard to compose together, quickly leading to code that is difficult to read and maintain (known as "Callback Hell").

關於回調地域相關說明參見如下連接

回調地域

The second approach (mentioned earlier), seeking more efficiency, can be a solution to the resource wasting problem. By writing asynchronous, non-blocking code, you let the execution switch to another active task using the same underlying resources and later come back to the current process when the asynchronous processing has finished.

But how can you produce asynchronous code on the JVM? Java offers two models of asynchronous programming:

• Callbacks: Asynchronous methods do not have a return value but take an extra callback parameter (a lambda or anonymous class) that gets called when the result is available. A well known example is Swing’s EventListenerhierarchy.
• Futures: Asynchronous methods return a Future<T> immediately. The asynchronous process computes a T value, but the Future object wraps access to it. The value is not immediately available, and the object can be polled until the value is available. For instance, ExecutorService running Callable<T> tasks use Future objects.

2. Reactive使用場景

Long Live模式：Netty I/O鏈接（RPC）

Short Live模式：HTTP(須要超時時間)

短平快的應用並不適合Reactive，須要作合理的timeout時間規劃

Reactive 性能測試： https://blog.ippon.tech/sprin...

3. Reactive理解誤區

Web：快速響應（Socket Connection Timeout）

Tomcat Connector Thread Pool（200個線程） --> Reactive Thread Pool(50個線程)

I/O鏈接從Tomcat轉移到了Reactive

• Tomcat Thread 負責處理鏈接（能夠由200個鏈接擴充到2000個鏈接）

• Reactive Thread負責處理任務（有可能會處理不過來）

  • 繼續等待（Timeout無限）
  • Timeout

4. 設計Reactive Web

4.1 異步設計

利用Servlet 3.1規範中的異步處理（Asynchronous processing），異步上下文（AsyncContext）實現異步Servlet。

Some times a filter and/or servlet is unableto complete the processing of a request
without waiting for a resource or event before generating a response. For example, a
servlet may need to wait for an available JDBC connection, for a response from a
remote web service, for a JMS message, or for an application event, before
proceeding to generate a response. Waiting within the servlet is an inefficient
operation as it is a blocking operation that consumes a thread and other limited
resources. Frequently a slow resource such as a database may have many threads
blocked waiting for access and can cause thread starvation and poor quality of
service for an entire web container.
The asynchronous processing of requests is introduced to allow the thread may
return to the container and perform other tasks. When asynchronous processing
begins on the request, another thread or callback may either generate the response
and call completeor dispatch the request so that it may run in the context of the
container using the AsyncContext.dispatchmethod. A typical sequence of events
for asynchronous processing is:

1. The request is received and passed via normal filters for authentication etc. to the
   servlet.
2. The servlet processes the request parameters and/or content to determine the
   nature of the request.
3. The servlet issues requests for resources or data, for example, sends a remote web
   service request or joins a queue waiting for a JDBC connection.
4. The servlet returns without generating a response.
5. After some time, the requested resourcebecomes available, the thread handling
   that event continues processing either in the same thread or by dispatching to a
   resource in the container using the AsyncContext.

4.2 非阻塞設計

利用Servlet 3.1規範中的Non Blocking IO

Non-blocking request processing in the Web Container helps improve the ever
increasing demand for improved Web Container scalability, increase the number of
connections that can simultaneously be handled by the Web Container. Nonblocking IO in the Servlet container allowsdevelopers to readdata as it becomes
available or write data when possible to do so. Non-blocking IO only works with
async request processing in Servlets and Filters (as defined in Section 2.3.3.3,
「Asynchronous processing」 on page 2-10), and upgrade processing (as defined in
Section 2.3.3.5, 「Upgrade Processing」 on page 2-20). Otherwise, an
IllegalStateExceptionmust be thrown when
ServletInputStream.setReadListeneror
ServletOutputStream.setWriteListeneris invoked.

5. 實現Reactive Web

實現代碼連接：github