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

程序员文章站 2022-06-20 10:35:49

前言什么是流式传输？流式传输是这一种以稳定持续流的形式传输数据的技术。流式传输的使用场景有些场景中，服务器返回的数据量较大，等待时间较长，客户端不得...

前言

什么是流式传输？

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

流式传输的使用场景

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

signalr

signalr是一个.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 SignalR中的流式传输深入讲解

选择创建asp.net core 2.1的web application

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

创建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个事件