什麼是流式傳輸？

流式傳輸是這一種以穩定持續流的形式傳輸資料的技術。

流式傳輸的使用場景

有些場景中，伺服器返回的資料量較大，等待時間較長，客戶端不得不等待伺服器返回所有資料後，再進行相應的操作。這時候使用流式傳輸，可以將伺服器資料碎片化，當每個資料碎片讀取完成之後，就只傳輸完成的部分，而不需要等待所有資料都讀取完成。

如何在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>\[email protected]\signalr\dist\browser目錄中拷貝到wwwroot\lib\signalr目錄下

編寫頁面

拷貝以下程式碼到Index.cshtml

@page
@model IndexModel
@{
    ViewData["Title"] = "Home page";
}

<div class="container">
    <div class="row">&nbsp;</div>
    <div class="row">
        <div class="col-6">&nbsp;</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">&nbsp;</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的流式傳輸改善使用者體驗，減輕伺服器壓力。