[[375247]]

一 名詞理解

1 流式

流式(Stream)亦稱響應式，是一種基于異步數據流研發框架，是一種概念和編程模型，并非一種技術架構，目前在各技術棧都有響應式的技術框架，前端的React.js、RxJs，服務端以RxJava、Reactor，Android端的RXJava。由此而來的即是響應式編程。

2 反應式/響應式編程

反應式編程/響應式編程(Reactive Programming)是一種基于事件模型編程范式，眾所周知異步編程模式中通常有兩種獲得上一個任務執行結果的方式，一個就是主動輪訓，我們把它稱為Proactive方式。另一個就是被動接收反饋，我們稱為Reactive。簡單來說，在Reactive方式中，上一個任務的結果的反饋就是一個事件，這個事件的到來將會觸發下一個任務的執行。

這也就是Reactive的內涵。我們把處理和發出事件的主體稱為Reactor，它可以接收事件并處理，也可以在處理完事件后，發出下一個事件給其他Reactor。

下面是一個Reactive模型的示意圖：

當然一種新的編碼模式，它的RunTime會減少上下文切流從而提升性能，減少內存消耗，與之相反帶來的是代碼的可維護性降低。衡量優劣需要根據場景帶來的收益來衡量。

3 流式輸出

流式輸出就比較神奇，源自于團隊內部在一次性能大賽結束后的總結中產生，是基于流式的理論基礎在頁面渲染以及渲染的HTML在網絡傳輸中的具體應用而誕生，也有人也簡單的稱之為流式渲染。即：將頁面拆分成獨立的幾部分模塊，每個模塊有單獨的數據源和單獨的頁面模板，在server端流式的操作每個模塊進行業務邏輯處理和頁面模板的渲染，然后流式的將渲染出來的HTML輸出到網絡中，接著分塊的HTML數據在網絡中傳輸，接著流式的分塊的HTML在瀏覽器逐個渲染展示。具體流程如下：

針對HTML可以如上所述進行流式輸出，衍生出針對json數據的流式輸出，其實也是如出一轍，無非少了一層渲染的邏輯，數據流式輸出流程跟上圖類似，不再贅述。這里可以把客戶端的請求當做響應式的一個事件，所以總結就是客戶端主動發出請求，服務端流式返回數據，即流式輸出。

4 端到端響應式

基于流式輸出，我們再深入一點，可以發現其實不只是用戶端和web server之間的數據可以在網絡上進行流式輸出，微服務的各個server之間的數據其實也可以在網絡上進行流式輸出，如下圖所示：

數據可以在網絡之間的流式傳輸，再進一步來看，數據在整條請求響應鏈路上的流式傳輸會是什么樣子，見下圖所示：

綜上所述我們定義：端到端響應式=流式輸出+響應式編程。

二 流式輸出理論基礎

是什么基礎技術理論，支撐我們能夠像上述流程那樣對數據進行流式輸出和接收，下面有幾個核心的技術點：

1 HTTP分塊傳輸協議

分塊傳輸編碼(Chunked transfer encoding)是超文本傳輸協議(HTTP)中的一種數據傳輸機制，允許HTTP由網頁服務器發送給客戶端應用( 通常是網頁瀏覽器)的數據可以分成多個部分。分塊傳輸編碼只在HTTP協議1.1版本(HTTP/1.1)中提供。

如果需要使用分塊傳輸編碼的響應格式，我們需要在HTTP響應中設置響應頭Transfer-Encoding: chunked。它的具體傳輸格式是這樣的(注意HTTP響應中換行符是\r\n)：

HTTP/1.1 200 OK\r\n 
\r\n 
Transfer-Encoding: chunked\r\n 
...\r\n 
\r\n 
<chunked 1 length>\r\n 
<chunked 1 content>\r\n 
<chunked 2 length>\r\n 
<chunked 2 content>\r\n 
...\r\n 
0\r\n 
\r\n 
\r\n 

具體流程見流式輸出名詞理解部分，分塊傳輸編碼例子：

func handleChunkedHttpResp(conn net.Conn) { 
    buffer := make([]byte, 1024) 
    n, err := conn.Read(buffer) 
    if err != nil { 
        log.Fatalln(err) 
    } 
    fmt.Println(n, string(buffer)) 
 
    conn.Write([]byte("HTTP/1.1 200 OK\r\n")) 
    conn.Write([]byte("Transfer-Encoding: chunked\r\n")) 
    conn.Write([]byte("\r\n")) 
 
    conn.Write([]byte("6\r\n")) 
    conn.Write([]byte("hello,\r\n")) 
 
    conn.Write([]byte("8\r\n")) 
    conn.Write([]byte("chunked!\r\n")) 
 
    conn.Write([]byte("0\r\n")) 
    conn.Write([]byte("\r\n")) 
} 

這里需要注意的是HTTP分塊傳輸對同步HTML輸出比較適合(對于瀏覽器來講)，因為在很多web頁面涉及SEO，SEO的TDK元素必須同步輸出，所以這種方式比較適合，針對于JSON數據的流式輸出通過SSE來實現，具體如下。

2 HTTP SSE協議

sse(Server Send Events)是HTTP的標準協議，是服務端向客戶端發送事件流式的方式。在客戶端中為一些事件類型綁定監聽函數，從而做業務邏輯處理。這里要注意的是SEE是單向的，只能服務器向客戶端發送事件流，具體流程如下：

SSE協議中約束了下面幾個字段類型

1)event

事件類型。如果指定了該字段,則在客戶端接收到該條消息時，會在當前的EventSource對象上觸發一個事件，事件類型就是該字段的字段值,你可以使用addEventListener()方法在當前EventSource對象上監聽任意類型的命名事件，如果該條消息沒有event字段，則會觸發onmessage屬性上的事件處理函數。

2)data

消息的數據字段。如果該條消息包含多個data字段,則客戶端會用換行符把它們連接成一個字符串來作為字段值。

3)id

事件ID，會成為當前EventSource對象的內部屬性"最后一個事件ID"的屬性值。

4)retry

一個整數值，指定了重新連接的時間(單位為毫秒)，如果該字段值不是整數,則會被忽略。

客戶端代碼示例

// 客戶端初始化事件源 
const evtSource = new EventSource("//api.example.com/ssedemo.php", { withCredentials: true } ); 
 
// 對 message 事件添加一個處理函數開始監聽從服務器發出的消息 
evtSource.onmessage = function(event) { 
  const newElement = document.createElement("li"); 
  const eventList = document.getElementById("list"); 
 
  newElement.innerHTML = "message: " + event.data; 
  eventList.appendChild(newElement); 
} 

服務器代碼示例

date_default_timezone_set("America/New_York"); 
header("Cache-Control: no-cache"); 
header("Content-Type: text/event-stream"); 
$counter = rand(1, 10); 
while (true) { 
  // Every second, send a "ping" event. 
  echo "event: ping\n"; 
  $curDate = date(DATE_ISO8601); 
  echo 'data: {"time": "' . $curDate . '"}'; 
  echo "\n\n"; 
  // Send a simple message at random intervals. 
  $counter--; 
  if (!$counter) { 
    echo 'data: This is a message at time ' . $curDate . "\n\n"; 
    $counter = rand(1, 10); 
  } 
  ob_end_flush(); 
  flush(); 
  sleep(1); 
} 

效果示例

event: userconnect 
data: {"username": "bobby", "time": "02:33:48"} 
event: usermessage 
data: {"username": "bobby", "time": "02:34:11", "text": "Hi everyone."} 
event: userdisconnect 
data: {"username": "bobby", "time": "02:34:23"} 
event: usermessage 
data: {"username": "sean", "time": "02:34:36", "text": "Bye, bobby."} 

這里需要注意下，在未通過http2使用SSE時，SSE會收到最大連接數限制，此時默認的最大連接數只有6，即同一時間只能建立6個SSE連接，不過這里的限制是對同域名的，跨域的域名可以再建立6個SSE連接。通過HTTP2使用SSE時默認的最大連接數是100。

目前SSE已集成到spring5，Springboot2的webflux其實就是通過SSE的方式進行數據的流式輸出。

3 WebSocket

Websocket就比較老生常談了，這里主要介紹下它與SSE的區別：

• Websocket是區別于HTTP的另外一種協議，是全雙工通信，協議相對來說比較中，對代碼侵入度比較高。
• SSE是標準的HTTP協議，是半雙工通信，支持斷線重連和自定義事件和數據類型，相對輕便靈活。

4 RSocket

在微服務架構中，不同服務之間通過應用協議進行數據傳輸。典型的傳輸方式包括基于 HTTP 協議的 REST 或 SOAP API 和基于 TCP 字節流的 RPC 等。但是對于HTTP只支持請求響應模式，如果客戶端需要獲取最新的推送消息，就必須使用輪詢，這無疑造成了大量的資源浪費。再者如果某個請求的響應時間過長，會阻塞之后的其他請求的處理;雖然服務器發送事件(Server-Sent Events，SSE)可以用來推送消息，不過 SSE 是一個簡單的文本協議，僅提供有限的功能;而WebSocket 可以進行雙向數據傳輸，不過它沒有提供應用層協議支持，Rsocket很好的解決了已有協議面臨的各種問題。

Rsocket是一個面向反應式應用程序的新型應用網絡協議，它工作在網絡七層模型中 5/6 層的協議，是 TCP/IP 之上的應用層協議，RSocket 可以使用不同的底層傳輸層，包括 TCP、WebSocket 和 Aeron。TCP 適用于分布式系統的各個組件之間交互，WebSocket 適用于瀏覽器和服務器之間的交互，Aeron 是基于 UDP 協議的傳輸方式，這就保證了 RSocket 可以適應于不同的場景，見上圖。然后RSocket 使用二進制格式，保證了傳輸的高效，節省帶寬。而且，通過基于反應式流控保證了消息傳輸中的雙方不會因為請求的壓力過大而崩潰。更多詳細資料請移步RSocket[1]。雷卷也開源了alibaba-rsocket-broker[2]，感興趣可以去深入了解請教。

Rsocket提供了四種不同的交互模式滿足所有場景：

RSocket 提供了不同語言的實現，包括Java、Kotlin、JavaScript、Go、.NET和C++ 等，如下為僅供學習了解的簡單Java實現：

import io.rsocket.AbstractRSocket; 
import io.rsocket.Payload; 
import io.rsocket.RSocket; 
import io.rsocket.RSocketFactory; 
import io.rsocket.transport.netty.client.TcpClientTransport; 
import io.rsocket.transport.netty.server.TcpServerTransport; 
import io.rsocket.util.DefaultPayload; 
import reactor.core.publisher.Mono; 
 
public class RequestResponseExample { 
 
  public static void main(String[] args) { 
    RSocketFactory.receive() 
        .acceptor(((setup, sendingSocket) -> Mono.just( 
            new AbstractRSocket() { 
              @Override 
              public Mono<Payload> requestResponse(Payload payload) { 
                return Mono.just(DefaultPayload.create("ECHO >> " + payload.getDataUtf8())); 
              } 
            } 
        ))) 
        .transport(TcpServerTransport.create("localhost", 7000)) //指定傳輸層實現 
        .start() //啟動服務器 
        .subscribe(); 
 
    RSocket socket = RSocketFactory.connect() 
        .transport(TcpClientTransport.create("localhost", 7000)) //指定傳輸層實現 
        .start() //啟動客戶端 
        .block(); 
 
    socket.requestResponse(DefaultPayload.create("hello")) 
        .map(Payload::getDataUtf8) 
        .doOnNext(System.out::println) 
        .block(); 
 
    socket.dispose(); 
  } 
} 

5 響應式編程框架

如果要在全鏈路實現響應式，那響應式編程框架是支撐這個技術的核心技術，這對于開發者來說是一種編程模式的變革，通過使用異步數據流進行編程對于原流程化的編程模式來說變化還很大。

簡單示例如下：

@Override 
public Single<Integer> remaining() { 
    return Flowable.fromIterable(LotteryEnum.EFFECTIVE_LOTTERY_TYPE_LIST) 
        .flatMap(lotteryType -> tairMCReactive.get(generateLotteryKey(lotteryType))) 
        .filter(Result::isSuccess) 
        .filter(result -> !ResultCode.DATANOTEXSITS.equals(result.getRc())) 
        .map(result -> (Integer) result.getValue().getValue()) 
        .reduce((acc, lotteryRemaining) -> acc + lotteryRemaining) 
        .toSingle(0); 
} 

總的來說通過HTTP分塊傳輸協議和HTTP SSE協議以及RSocket我們可以實現流式輸出，通過流式輸出和響應式編程端到端的響應式才得以實現。

三 流式輸出應用場景

性能、體驗和數據是我們日常工作中抓的最緊的三件事情。對于性能來說也一直是我們追求極致和永無止境的核心點，流式輸出也是在解決性能體驗這個問題而誕生，那是不是所有的場景都適合流式輸出呢?當然不是，我們來康康哪些場景適合?

以上為Resource Timing API規范提供的請求生命周期包含的主要階段，通過上述來看下一下幾個場景對于請求生命周期的影響。

1 頁面流式輸出場景

對于動態頁面來說(相對于靜態頁面)主要由頁面樣式、頁面交互的JS以及頁面的動態數據構成，除了上述請求生命周期的各階段耗時，還有頁面渲染耗時階段。瀏覽器拿到HTML會先進行DOM樹構建、預加載掃描器、CSSOM樹構建，Javascript編譯執行，在過程中CSS文件的加載和JS文件的加載阻塞頁面渲染過程。如果我們將頁面按照以下方式進行拆分進行流式輸出將會在性能上有很大的收益。

單接口動態頁面

對于某些場景比如SEO，頁面需要同步渲染輸出，此時頁面通常是單接口動態頁面，就可以將頁面拆分成body以上部分和body以下的部分，例如：

<!-- 模塊1 --> 
<html> 
  <head> 
  <meta  /> 
    <link  />   
  <style></style>   
  <script src=""></script> 
  </head> 
  <body> 
 
<!-- 模塊2 --> 
        <div>xxx</div> 
        <div>yyy</div>   
        <div>zzz</div>  
    </body> 
</html>   

當模塊1到達頁面模塊2未到達時，模塊1渲染后在等待模塊2到來的同時可以進行CSS和JS的加載，在幾個方面進行了性能提升：

• 到達瀏覽器的首字節時間(TTFB)
• 數據包到達瀏覽器下載HTML的時間
• CSS和JS的加載及執行時間
• 拆成模塊之后網絡傳輸的時間會有一定的降低

單接口多樓層頁面

<!-- 模塊1 --> 
<html> 
  <head> 
  <meta  /> 
    <link  />   
  <style></style>   
  <script src=""></script> 
  </head> 
  <body> 
 
<!-- 模塊2 --> 
        <div>xxx1</div> 
        <div>yyy1</div>   
        <div>zzz1</div>  
     
<!-- 模塊3 --> 
        <div>xxx2</div> 
        <div>yyy2</div>   
        <div>zzz2</div> 
     
<!-- 模塊4 --> 
        <div>xxx3</div> 
        <div>yyy3</div>   
        <div>zzz3</div> 
    </body> 
</html>   

很多場景是一個頁面展現多個樓層、譬如首頁的四大金剛以及各種導購樓層，detail的信息樓層等等，甚至后面樓層依賴前面樓層的數據，類似這種情況可以將頁面樓層拆分成多個模塊進行輸出，在上述幾個方面進行了性能提升之外還有額外的性能提升：樓層之間數據相互依賴的數據處理時間。

多接口多樓層頁面

一般情況下大部分頁面都是由同步SSR渲染和異步CSR渲染進行，這時會涉及到JS異步加載異步樓層，如果同步渲染部分按照單接口多樓層進行拆分會在上述基礎上提前加載運行異步樓層的渲染。

總的來說基于HTTP分塊傳輸協議的流式輸出幾乎覆蓋所有頁面場景，供所有頁面提升性能體驗。

2 數據流式輸出場景

單接口大數據

對于APP或者單頁面系統更多的是通過異步加載數據的方式進行頁面渲染，單個接口會造成單個接口的RT時間較長，以及數據包體太大導致在網絡中拆包粘包的損耗較大。如果通過多個異步接口會因網絡帶寬受限而導致數據請求的延時較高以及網絡IO的帶來的CPU占有率較高，因此可以通過業務場景進行分析將單接口拆分成多個相互獨立或者有一定耦合關系的業務模塊，將這些模塊的數據進行流式輸出，以此帶來以下性能體驗上的提升。

• 到達瀏覽器的首字節時間(TTFB)
• 數據包到達端側下載數據的時間
• 數據在網絡傳輸的時間

多相互依賴接口

但是在大部分場景中我們遇到的業務場景是相互耦合關聯的，比方說榜單模塊數據依賴它上面的新品模塊的數據進行業務邏輯處理，這種情況在服務器側處理完新品模塊數據后對數據進行輸出，再接著處理榜單模塊數據進行輸出，這里接節省了相互依賴等待的時間。

當然日常的業務場景會相對復雜的多，但是通過流式輸出都會頁面性能和體驗會有很大的提升和助力。

四 小結

• 流式輸出的前世為流式渲染，今生為端到端的響應式，這些雖然帶來了性能體驗上的提升，但對研發模式變革的接受程度和運維成本的增加需要加以權衡。
• 簡單介紹了幾種流式輸出的技術方案，適合不同的業務場景。
• 提出了流式輸出適合的幾種場景，以及對頁面和數據進行拆分的方法。

相關鏈接

[1]https://rsocket.io/

[2]https://github.com/alibaba/alibaba-rsocket-broker