什么是流式傳輸?
流式傳輸是這一種以穩定持續流的形式傳輸數據的技術。
流式傳輸的使用場景
有些場景中,服務器返回的數據量較大,等待時間較長,客戶端不得不等待服務器返回所有數據后,再進行相應的操作。這時候使用流式傳輸,可以將服務器數據碎片化,當每個數據碎片讀取完成之后,就只傳輸完成的部分,而不需要等待所有數據都讀取完成。
如何在ASP.NET Core SignalR中啟用流式傳輸
在ASP.NET Core SignalR中當一個Hub
方法的返回值是ChannelReader
或者Task<ChannelReader
, 這個Hub
方法自動就會變成一個流式傳輸Hub
方法。
下面我們來做了一個簡單的例子
創建一個ASP.NET Core Web應用
首先我們使用Visual Studio 2017創建一個ASP.NET Core Web應用程序。
選擇創建ASP.NET Core 2.1的Web Application
創建Hub
下面我們添加一個StreamHub
類,代碼如下
public class StreamHub : Hub
{
public ChannelReader<int> DelayCounter(int delay)
{
var channel = Channel.CreateUnbounded<int>();
_ = WriteItems(channel.Writer, 20, delay);
return channel.Reader;
}
private async Task WriteItems(ChannelWriter<int> writer, int count, int delay)
{
for (var i = 0; i < count; i++)
{
await writer.WriteAsync(i);
await Task.Delay(delay);
}
writer.TryComplete();
}
}
DelayCounter
是一個流式傳輸方法, 它定義了一個延遲參數delay
, 定義了推送數據碎片的間隔時間WriteItems
是一個私有方法,它返回了一個Task
對象WriteItems
方法的最后一行writer.TryComplete()
表明了流式傳輸完成
配置SignalR
首先我們在Startup
類的ConfigureService
方法中添加SignalR服務
services.AddSignalR();
然后我們還需要為SignalR流添加路由,我們需要在Startup
類的Configure
方法中添加如下代碼:
app.UseSignalR(routes =>
{
routes.MapHub<StreamHub>("/streamHub");
});
添加SignalR客戶端腳本庫
這一步中我們需要在客戶端中添加SignalR JS庫。
這里我們需要借助npm來下載SignalR JS庫。
npm install @aspnet/signalr
安裝完成后,我們手動將signalr.js從<projectfolder>\node_modules@aspnet\signalr\dist\browser
目錄中拷貝到wwwroot\lib\signalr
目錄下
編寫頁面
拷貝以下代碼到Index.cshtml
@page
@model IndexModel
@{
ViewData["Title"] = "Home page";
}
<div class="container">
<div class="row"> </div>
<div class="row">
<div class="col-6"> </div>
<div class="col-6">
<input type="button" id="streamButton" value="Start Streaming" />
</div>
</div>
<div class="row">
<div class="col-12">
<hr />
</div>
</div>
<div class="row">
<div class="col-6"> </div>
<div class="col-6">
<ul id="messagesList"></ul>
</div>
</div>
</div>
<script src="~/lib/signalr/signalr.js"></script>
<script src="~/js/signalrstream.js"></script>
JavaScript中啟用流式傳輸
在wwwroot\js
目錄中創建一個新文件signalrstream.js
,代碼如下
var __awaiter = (this && this.__awaiter) || function (thisArg, _arguments, P, generator) {
return new (P || (P = Promise))(function (resolve, reject) {
function fulfilled(value) { try { step(generator.next(value)); } catch (e) { reject(e); } }
function rejected(value) { try { step(generator["throw"](value)); } catch (e) { reject(e); } }
function step(result) { result.done ? resolve(result.value) : new P(function (resolve) { resolve(result.value); }).then(fulfilled, rejected); }
step((generator = generator.apply(thisArg, _arguments || [])).next());
});
};
var connection = new signalR.HubConnectionBuilder()
.withUrl("/streamHub")
.build();
document.getElementById("streamButton").addEventListener("click", (event) => __awaiter(this, void 0, void 0, function* () {
try {
connection.stream("DelayCounter", 500)
.subscribe({
next: (item) => {
var li = document.createElement("li");
li.textContent = item;
document.getElementById("messagesList").appendChild(li);
},
complete: () => {
var li = document.createElement("li");
li.textContent = "Stream completed";
document.getElementById("messagesList").appendChild(li);
},
error: (err) => {
var li = document.createElement("li");
li.textContent = err;
document.getElementById("messagesList").appendChild(li);
},
});
}
catch (e) {
console.error(e.toString());
}
event.preventDefault();
}));
(() => __awaiter(this, void 0, void 0, function* () {
try {
yield connection.start();
}
catch (e) {
console.error(e.toString());
}
}))();
代碼解釋
與傳統SignalR不同,這里我們使用了不同的語法創建一個SignalR連接
var connection = new signalR.HubConnectionBuilder()
.withUrl("/streamHub")
.build();
對於一般的SignalR連接,我們會使用connection.on
方法來添加監聽器,但是在使用流式傳輸的時候,我們需要改用connection.stream
方法, 這個方法有2個參數
- Hub方法名稱, 本例中是
DelayCounter
- Hub方法的參數, 本例中是
500
connection.stream("DelayCounter", 500)
.subscribe({
next: (item) => {
var li = document.createElement("li");
li.textContent = item;
document.getElementById("messagesList").appendChild(li);
},
complete: () => {
var li = document.createElement("li");
li.textContent = "Stream completed";
document.getElementById("messagesList").appendChild(li);
},
error: (err) => {
var li = document.createElement("li");
li.textContent = err;
document.getElementById("messagesList").appendChild(li);
},
});
connection.stream
方法的返回對象中有一個subscribe
方法,這個方法中可以注冊3個事件
- next - 獲得到一個數據碎片時執行
- complete - 流式傳輸完成時執行
- error - 流式傳輸異常時執行
最終效果
總結
流式傳輸不是一個新概念,但是對ASP.NET Core SignalR來說,這是一個非常棒的特性。流式傳輸保證的用戶體驗的流暢,也降低了服務器壓力。
大部分程序員都知道SignalR不能傳輸過大的數據,但是使用流式傳輸之后,客戶端不需要一次性等待服務器端返回所有數據,所以如果你的項目單次請求的數據量很大,可以考慮使用SignalR的流式傳輸改善用戶體驗,減輕服務器壓力。