ASP.NET Core SignalR中的流式传输深入讲解

前言什么是流式传输？流式传输是这一种以稳定持续流的形式传输数据的技术。流式传输的使用场景有些场景中，服务器返回的数据量较大，等待时间较长，客户端不得不等待服务器返回所有数据后，再进行相应的操作。这时候使用流式传输，可以将服务器数据碎片化，当每个数据碎片读取完成之后，就只传输完成的部分，而不需要等待所有数据都读取完成。
SignalRSignalR是一个。NET Core/。NET Framework的开源实时框架。 SignalR的可使用Web Socket, Server Sent Events 和 Long Polling作为底层传输方式。SignalR基于这三种技术构建, 抽象于它们之上, 它让你更好的关注业务问题而不是底层传输技术问题。
SignalR这个框架分服务器端和客户端, 服务器端支持ASP。NET Core 和 ASP。NET; 而客户端除了支持浏览器里的javascript以外, 也支持其它类型的客户端, 例如桌面应用。如何在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>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方法的参数, 本例中是500connection。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 – 流式传输异常时执行最终效果。