流式传输是这一种以稳定持续流的形式传输数据的技术。

有些场景中，服务器返回的数据量较大，等待时间较长，客户端不得不等待服务器返回所有数据后，再进行相应的操作。这时候使用流式传输，可以将服务器数据碎片化，当每个数据碎片读取完成之后，就只传输完成的部分，而不需要等待所有数据都读取完成。

在ASP.NET Core SignalR中当一个Hub方法的返回值是ChannelReader或者Task<ChannelReader>, 这个Hub方法自动就会变成一个流式传输Hub方法。

下面我们来做了一个简单的例子

首先我们使用Visual Studio 2017创建一个ASP.NET Core Web应用程序。

选择创建ASP.NET Core 2.1的Web Application

下面我们添加一个StreamHub类，代码如下

  1. public class StreamHub : Hub
  2. {   
  3.     public ChannelReader<int> DelayCounter(int delay)
  4.     {
  5.         var channel = Channel.CreateUnbounded<int>();
  7.         _ = WriteItems(channel.Writer, 20, delay);
  9.         return channel.Reader;
  10.     }
  12.     private async Task WriteItems(ChannelWriter<int> writer, int count, int delay)
  13.     {
  14.         for (var i = 0; i < count; i++)
  15.         {
  16.             await writer.WriteAsync(i);
  17.             await Task.Delay(delay);
  18.         }
  20.         writer.TryComplete();
  21.     }
  22. }
  • DelayCounter是一个流式传输方法, 它定义了一个延迟参数delay, 定义了推送数据碎片的间隔时间
  • WriteItems是一个私有方法，它返回了一个Task对象
  • WriteItems方法的最后一行writer.TryComplete()表明了流式传输完成

首先我们在Startup类的ConfigureService方法中添加SignalR服务

  1. services.AddSignalR();

然后我们还需要为SignalR流添加路由，我们需要在Startup类的Configure方法中添加如下代码：

  1. app.UseSignalR(routes =>
  2. {
  3.    routes.MapHub<StreamHub>("/streamHub");
  4. });

这一步中我们需要在客户端中添加SignalR JS库。

这里我们需要借助npm来下载SignalR JS库。

  1. npm install @aspnet/signalr

安装完成后，我们手动将signalr.js从<projectfolder>\node_modules@aspnet\signalr\dist\browser目录中拷贝到wwwroot\lib\signalr目录下

拷贝以下代码到Index.cshtml

  1. @page
  2. @model IndexModel
  3. @{
  4.     ViewData["Title"] = "Home page";
  5. }
  7. <div class="container">
  8.     <div class="row">&nbsp;</div>
  9.     <div class="row">
  10.         <div class="col-6">&nbsp;</div>
  11.         <div class="col-6">
  12.             <input type="button" id="streamButton" value="Start Streaming" />
  13.         </div>
  14.     </div>
  15.     <div class="row">
  16.         <div class="col-12">
  17.             <hr />
  18.         </div>
  19.     </div>
  20.     <div class="row">
  21.         <div class="col-6">&nbsp;</div>
  22.         <div class="col-6">
  23.             <ul id="messagesList"></ul>
  24.         </div>
  25.     </div>
  26. </div>
  27. <script src="~/lib/signalr/signalr.js"></script>
  28. <script src="~/js/signalrstream.js"></script>

在wwwroot\js目录中创建一个新文件signalrstream.js ,代码如下

  1. var __awaiter = (this && this.__awaiter) || function (thisArg, _arguments, P, generator) {
  2.     return new (P || (P = Promise))(function (resolve, reject) {
  3.         function fulfilled(value) { try { step(generator.next(value)); } catch (e) { reject(e); } }
  4.         function rejected(value) { try { step(generator["throw"](value)); } catch (e) { reject(e); } }
  5.         function step(result) { result.done ? resolve(result.value) : new P(function (resolve) { resolve(result.value); }).then(fulfilled, rejected); }
  6.         step((generator = generator.apply(thisArg, _arguments || [])).next());
  7.     });
  8. };
  10. var connection = new signalR.HubConnectionBuilder()
  11.     .withUrl("/streamHub")
  12.     .build();
  13. document.getElementById("streamButton").addEventListener("click", (event) => __awaiter(this, void 0, void 0, function* () {
  14.     try {
  15.         connection.stream("DelayCounter", 500)
  16.             .subscribe({
  17.                 next: (item) => {
  18.                     var li = document.createElement("li");
  19.                     li.textContent = item;
  20.                     document.getElementById("messagesList").appendChild(li);
  21.                 },
  22.                 complete: () => {
  23.                     var li = document.createElement("li");
  24.                     li.textContent = "Stream completed";
  25.                     document.getElementById("messagesList").appendChild(li);
  26.                 },
  27.                 error: (err) => {
  28.                     var li = document.createElement("li");
  29.                     li.textContent = err;
  30.                     document.getElementById("messagesList").appendChild(li);
  31.                 },
  32.             });
  33.     }
  34.     catch (e) {
  35.         console.error(e.toString());
  36.     }
  37.     event.preventDefault();
  38. }));
  40. (() => __awaiter(this, void 0, void 0, function* () {
  41.     try {
  42.         yield connection.start();
  43.     }
  44.     catch (e) {
  45.         console.error(e.toString());
  46.     }
  47. }))();

与传统SignalR不同，这里我们使用了不同的语法创建一个SignalR连接

  1. var connection = new signalR.HubConnectionBuilder()
  2.     .withUrl("/streamHub")
  3.     .build();

对于一般的SignalR连接，我们会使用connection.on方法来添加监听器，但是在使用流式传输的时候，我们需要改用connection.stream方法, 这个方法有2个参数

  • Hub方法名称, 本例中是DelayCounter
  • Hub方法的参数, 本例中是500 
  1. connection.stream("DelayCounter", 500)
  2.     .subscribe({
  3.         next: (item) => {
  4.             var li = document.createElement("li");
  5.             li.textContent = item;
  6.             document.getElementById("messagesList").appendChild(li);
  7.         },
  8.         complete: () => {
  9.             var li = document.createElement("li");
  10.             li.textContent = "Stream completed";
  11.             document.getElementById("messagesList").appendChild(li);
  12.         },
  13.         error: (err) => {
  14.             var li = document.createElement("li");
  15.             li.textContent = err;
  16.             document.getElementById("messagesList").appendChild(li);
  17.         },
  18. });

connection.stream方法的返回对象中有一个subscribe方法，这个方法中可以注册3个事件

  • next – 获得到一个数据碎片时执行
  • complete – 流式传输完成时执行
  • error – 流式传输异常时执行

流式传输不是一个新概念，但是对ASP.NET Core SignalR来说，这是一个非常棒的特性。流式传输保证的用户体验的流畅，也降低了服务器压力。

大部分程序员都知道SignalR不能传输过大的数据，但是使用流式传输之后，客户端不需要一次性等待服务器端返回所有数据，所以如果你的项目单次请求的数据量很大，可以考虑使用SignalR的流式传输改善用户体验，减轻服务器压力。

本篇源代码地址 https://github.com/lamondlu/StreamingInSignalR

版权声明：本文为lwqlun原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。
本文链接：https://www.cnblogs.com/lwqlun/p/9839305.html