用Java構建反應式REST API - Kalpa Senanayake

本文的重點是使用Java構建RESTFul API，同時受益於反應式程式設計模型。但與大多數關於此主題的其他文章不同，本文不會急於直接編寫程式碼。它將指導您完成此程式設計範例的主幹，以便您對其有充分的瞭解。然後使用該知識構建API。

該系列由兩部分組成。第一部分介紹了反應系統和反應式程式設計，並清除了這些術語之間的混淆。

然後介紹了反應式程式設計的基礎知識，並將傳統的併發模型與訊息/事件驅動的併發性進行了比較。

第二部分是關於使用Spring WebFlux來弄髒和構建RESTFul API，並向讀者介紹第四代反應框架。

反應系統

反應式系統架構是構建響應式系統的架構風格。這是在反應性宣言中定義的，我們將簡要介紹宣言中的每個關鍵專案，同時使用日常系統行為解釋其含義。

反應系統響應迅速：系統及時響應，提供一致的服務質量。

這意味著當負載高和低時系統以一致的方式執行。結果是使用者開始建立您的系統的信心，並繼續與系統做生意。

反應系統具有彈性：系統在出現故障時保持響應。

這意味著系統能夠隔離故障，包含故障並在必要時使用複製來緩解故障並繼續為使用者提供服務。

事情經常出錯。但是，如果我們擁有一個具有彈性的系統，那麼它就能變得如此敏捷。

反應系統具有彈性：系統在不同的工作負載下保持響應。

這意味著系統可以對負載的變化做出反應。這再次連結到響應性，因為您可以看到沒有彈性和彈性就無法實現響應性。

反應系統是訊息驅動的：系統依賴於非同步訊息傳遞。它將失敗作為訊息傳遞。並且它在必要時應用反壓來控制訊息流。

這意味著系統對事件/訊息作出反應，並且僅在資源處於活動狀態時消耗資源。

反應系統和反應式程式設計是兩回事。一種是架構風格，另一種是程式設計範例，可用於實現反應系統的某些特徵，但不是全部。

反應式程式設計

現在，我們對反應系統有了很好的理解。是時候讓我們深入瞭解反應式程式設計概念。

反應式程式設計是非同步程式設計範例的一個分支，它允許資訊驅動邏輯而不是依賴於執行的執行緒

現在，這聽起來像維基百科或一些學術研究論文。它丟擲的術語似乎有些可怕。用簡單的語言：

反應式程式設計允許應用程式基於訊息/事件在任意時間發生操作，而不是執行驅動的執行緒。

現在您可以看到訊息驅動和響應特性可以從響應式程式設計中受益。但並非所有，要實現所有這些目標，我們需要更廣泛的工具範圍。這些不屬於本文的範圍。

事件、事件、還是事件

不同型別的事件：

• 點選流：點擊發生在時間線上的不同時間點，它是一系列事件。當它發生時，無法保證它是適當的間隔。
• 新聞應用的通知：新聞專案會在任意時間內顯示在手機螢幕上作為推送通知。沒有人知道突發新聞何時發生。
• 來自遠端端點的HTTP響應：網路有自己的問題，如延遲和連線失敗，因此響應以任意時間間隔到達。

反應式程式設計和事件

這些型別的事件的共同因素是這些事件在任意時間發生，因此如果我們在程式中有一個執行執行緒，那些執行緒必須等待這些事件完成。

如果系統有更多客戶端正在等待這些操作的結果，那麼我們需要更多執行緒來伺服器客戶端。

這就是反應式程式設計脫穎而出的地方。它不是等待這些事件完成，而是提供類似觀察器的機制，讓這些事件驅動執行。

結果是處理大量這些事件的執行緒和資源數量減少。

我們需要了解的最重要的事實是：

反應式程式設計不會使應用程式更快，但它允許應用程式以更少的資源為更多客戶端提供服務。

如果我們需要擴充套件，我們可以橫向擴充套件（使用更少的資源），這使得它成為構建現代微服務的完美範例。這就是反應式程式設計如何加強反應系統的彈性特性。

反應式程式設計庫的特點

瞭解反應式程式設計的特徵是解開可能性的關鍵：

1. 執行是非同步和非阻塞的。這意味著呼叫執行緒不會阻塞I / O事件並等待它完成。我們將在本文後面詳細討論這個問題。
2. 無阻塞背壓。非阻塞背壓是一種允許事件訂閱者的方式以非阻塞方式控制事件流速率的機制。在阻塞情況下，會阻止釋出者，迫使釋出者等待消費者從堵塞方式恢復過來。
3. 這允許在緩慢的釋出者/快速接收者和快速釋出者/慢接收者的場景中巧妙地處理。
4. 支援反應流：一個無限制的事件/訊息流，在元件之間非同步傳遞元素，具有強制性非阻塞壓力。

結合上述所有內容，我們可以很好地瞭解使用反應式程式設計原理開發的應用程式。

這些應用程式支援處理無限數量的事件，事件驅動和了解他們正在處理的環境，並可以對這些環境中的更改做出反應。

為什麼它很重要，有什麼好處，還不清楚？

讓我們深入瞭解更多細節並進行討論。

我記得有一次我和其他開發人員談論過咖啡的反應性程式設計，他的問題是。

“使用它有什麼好處？”

“與我們今天使用的相比，它給桌面帶來了哪些好處？”

這些都是關於任何新技術的完全有效的問題。為了回答這些問題，我們需要退一步思考我們用於使用Java構建應用程式的工具。

用Java編寫的Web API通常部署到像Tomcat，WebSphere等servlet容器中。這些容器使用Servlet API進行操作，這些操作提供阻塞I / O. 因此流行的框架如Spring，Spring Web MVC也繼承了這種阻塞行為。

資料庫操作阻止I / O呼叫，JPA，JDBC都以這種方式執行。

這些阻塞操作會阻止請求執行緒，直到該操作完成。因此，更多請求會導致更多阻塞的執行緒等待I / O操作完成。結果如下。

1. 為執行緒之間的上下文切換支付更多的CPU時間。
2. 系統必須分配更多記憶體以支援越來越多的執行緒和這些執行緒的執行堆疊。
3. 更多記憶體意味著更多的GC時間和CPU上的GC開銷。

它不僅僅是I / O操作，Java併發工具的普通公民也擁有阻塞行為：

1. java.util.concurrent.Future，我們可以使用Future來表示非同步計算的結果。

   FutureTask<String> future =new FutureTask<String>(new Callable<String>() {public String call() {return searcher.search(target);}});executor.execute(future);

   但是當你需要結果時我們必須呼叫：

   //Waits if necessary for the computation to complete, and then retrieves its result.String result = future.get();

2. 同步synchronized 方法還強制執行緒在進入邏輯塊之前停止並檢查。

阻塞I / O方法的問題在於，使用阻塞API，我們無法支援反應系統。它是執行驅動，同步的執行緒。這兩個原因使得資源消耗很大。下一步將是找到一種方法來找到執行操作非同步非阻塞方式的更有效方法。

Java 8帶來了CompletableFuture，它是非同步程式設計的真正非同步非阻塞功能。但是如果你想要編寫更多結果並進行流處理，程式碼就會變得難以閱讀，並且缺乏流暢的操作API。

非阻塞方法

Java稍後引入了java.nio包，通過引入一個概念呼叫Selector來解決這種阻塞行為，它可以監視多個通道。

這允許單執行緒監視許多輸入通道。以及將資料載入到ByteBuffers而不是阻塞的Streams的關鍵概念。因此，ByteBuffers將提供可用的資料。

以下是使用NIO功能的伺服器演示。它是使用NIO實現的echo伺服器的簡單實現，但是足以獲得非阻塞方法基礎的示例。

import java.io.IOException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;

public class NIOServer {
private static final int PORT = 8888;
private static final int BUFFER_SIZE = 1024;
private static Selector selector = null;

public static void main(String[] args) {
logger("Starting NIOServer");
try {
InetAddress hostIP = InetAddress.getLocalHost();
logger(String.format("Trying to accept connections on %s:%d", hostIP.getHostAddress(), PORT));
// create selector via open();
selector = Selector.open();
// create a server socket channel
ServerSocketChannel server = ServerSocketChannel.open();
// get the server socket
ServerSocket serverSocket = server.socket();
InetSocketAddress address = new InetSocketAddress(hostIP, PORT);
// bind the server socket to address
serverSocket.bind(address);
// configure socket to be non-blocking
server.configureBlocking(false);
// register selector interest for accept event.
server.register(selector, SelectionKey.OP_ACCEPT);

while (true) {
// get a channel from selector, this will block until a channel get selected
selector.select();
// get keys for that channel
Set<SelectionKey> selectedKeys = selector.selectedKeys();
Iterator<SelectionKey> i = selectedKeys.iterator();

// go through selection keys one by one and see any of those events are ready
// if ready process that
while (i.hasNext()) {
SelectionKey key = i.next();

if (key.isAcceptable()) {

processAcceptEvent(server, key);

} else if (key.isReadable()) {

processReadEvent(key);
}
i.remove();
}
}
} catch (IOException e) {
logger(e.getMessage());
e.printStackTrace();
}
}

/**
* Handle the accept event
*
* @param socketServer socket channel
* @param keySelection key
* @throws IOExceptionIn case of error while accept the connection
*/<font>
<b>private</b> <b>static</b> <b>void</b> processAcceptEvent(ServerSocketChannel socket, SelectionKey key) throws IOException {
logger(</font><font>"Connection Accepted"</font><font>);
</font><font><i>// Accept the connection and make it non-blocking</i></font><font>
SocketChannel socketChannel = socket.accept();
socketChannel.configureBlocking(false);
</font><font><i>// Register interest in reading this channel</i></font><font>
socketChannel.register(selector, SelectionKey.OP_READ);
}

</font><font><i>/**
* Handle the read event
*
* @param keySelection key for the channel.
* @throws IOException
*/</i></font><font>
<b>private</b> <b>static</b> <b>void</b> processReadEvent(SelectionKey key) throws IOException {
logger(</font><font>"Handling ReadEvent"</font><font>);
</font><font><i>// create a ServerSocketChannel to read the request</i></font><font>
SocketChannel client = (SocketChannel) key.channel();
</font>

這種方法背後的基本原理是，Selector可以在多個通道中註冊它的興趣，當這些事件發生時，主執行緒通過呼叫匹配的處理邏輯來響應這些事件。

唯一的阻塞程式碼是第39行:

<font><i>//從選擇器獲取一個通道，這將阻塞直到一個通道被選中</i></font><font>
selector.select();
</font>

select（）方法阻塞，直到選擇一個通道。例如，直到發生新連線。

事件迴圈

上面的模式等於我們在JavaScript世界中稱為事件迴圈event loop的模式。Javascript是單執行緒執行時，因此它必須找到支援多個任務的方法，而不必建立多個執行緒。

當NodeJS出現並開始以較少的記憶體佔用和CPU時間來處理繁重的負載時，Java社群意識到這是解決這一系列問題的更具可擴充套件性的方法。眾所周知，多執行緒應用程式難以開發，難以維護。

反應性庫包

現在我們已經很好地理解了舊Java世界的同步阻塞行為以及使用事件迴圈進行非阻塞的新方法，我們可以開始進入Reactive世界。

首先，Microsoft為.NET框架建立了反應式擴充套件。並通過JavaScript跟進單執行緒，非阻塞，非同步語言，它對反應庫有真正的需求，因此RxJ就存在了。

• 反應庫的javaScript實現是RxJS
• 基於Netflix完成的RxJava實現

您可以在大多數流行的程式語言中找到Rx庫。反應式程式設計的當前庫提供以下內容。

1. 資料可用時，非阻塞操作的完整管道線。
2. 豐富的操作符集來操作這些事件流。
3. 背壓，控制生產者事件排放率的能力。
4. 能夠在程式碼中以良好的可讀性編排多個非同步任務。