# node 服务端主动推送数据

HTTP 协议遵循经典的客户端-服务器模型，客户端发送一个请求，然后等待服务器端的响应，服务器端只能在接收到客户端的请求之后进行响应，不能主动的发送数据到客户端。

客户端想要在不刷新页面的情况下实时获取到服务器端最新的数据，可以通过以下途径：

轮询
长轮询
HTTP 流
SSE
Web Sockets

# 1、轮询

轮询是在特定的的时间间隔（如每 1 秒），由浏览器对服务器发出 HTTP request，然后由服务器返回最新的数据给客户端的浏览器。这种传统的 HTTP request 的模式带来很明显的缺点 – 浏览器需要不断的向服务器发出请求，然而 HTTP request 的 header 是非常长的，里面包含的有用数据可能只是一个很小的值，这样会占用很多的带宽。

客户端（浏览器）定时向服务器端发送请求，获取最新的数据。可以通过在一个定时器中触发 ajax 请求来实现：

function showUnreadNews() {
  $(document).ready(function () {
    $.ajax({
      type: 'POST',
      url: 'unread_list.js',
      dataType: 'json',
      success: function (data) {
        alert(data);
      },
    });
  });
}
setInterval('showUnreadNews()', 5000); //轮询执行，5000ms一次

1
2
3
4
5
6
7
8
9
10
11
12
13

优点：

实现简单，JS 端进行一些更改即可，无需后端服务任何改动

缺点：

轮询的间隔过长，会导致用户不能及时接收到更新的数据；轮询的间隔过短，会导致查询请求过多，增加服务器端的负担

# 2、长轮询

长轮询方法实现原理如下：

客户端发起一个请求到服务器端(http request)
服务器端一直保持连接打开，直到有数据数据可发送给客户端，再返回这个请求(http response)
客户端收到服务器端返回的数据后，处理数据，并立马发起一个新的请求

// 客户端示例js
function showUnreadNews() {
  $.ajax({
    type: 'POST',
    url: 'unread_list.js',
    dataType: 'json',
    success: function (data) {
      //处理返回数据
      alert(data);
      //再次请求
      showUnreadNews();
    },
    complete: function (XMLHttpRequest, textStatus) {
      if (textStatus == 'timeout') {
        //判断是否超时
        showUnreadNews(); //超时，重新请求
      }
    },
  });
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20

// server端示例(nodejs)
var aTargets = [];
app.get('/notification', function(req, res) {
    aTargets.push(res);
  //res.end();  这里不调用res.end(),让http request连接一直存活着
})//此方法会在有新的数据时调用
onNewNotification : function (data) {
  aTargets.forEach(function(res){
    res.send(data);//当有新的数据时，再调用res.send(data)返回最新的数据,结束一次http请求
  })
}

1
2
3
4
5
6
7
8
9
10
11

优点：

可以及时获取到最新的数据
相较于轮询策略，减少了请求数量

缺点：

服务器端要一直保持连接，不能释放，由于一个服务器能够处理的连接数有限，当达到服务器处理的上限的时候，服务器将无法响应新的请求

# 3、HTTP 流

HTTP 流区别于轮询和长轮询方法，它在客户端网页的生命周期内，只需要使用一个 HTTP 连接，也就是只会向服务器发送一个请求，对于这个请求，服务器会保持 HTTP 连接（不返回 response），然后周期性的向浏览器发送数据。

首先我们创建一个服务器：

const express = require('express');
const app = express();
const port = 3100;

app.listen(port, () => {
  console.log('FastTest Admin API at: ', port);
});

1
2
3
4
5
6
7

然后，我们在需要流式返回响应的接口里设置相应头 Content-type: application/octet-stream。接下来，只要我们不断向输出流写入内容就可以了。哦，对了，结束的时候，我们还要关闭输出流。

// server端示例(nodejs)
const gzip = require('zlib').createGzip();
const fs = require('fs');

app.post('/httpstream', async (req, res, next) => {
  res.setHeader('Content-type', 'application/octet-stream');

  // 第一种输出流方式：调用 write 输出
  var x = 0;
  setInterval(() => {
    res.write(x++ + '|');
  }, 2000);
  // 关闭输出流
  // res.end();

  // 第二种输出流方式：从一个流转向另一个流，也就是数据本身就是一个流，然后调用 pipe 返回这个流
  const response = await apiTest(params); // apiTest 是一个 axios 请求方法，返回的也是一个流
  const stream = response.data;
  stream.pipe(res);

  // 第三种输出流方式：从文件转换为流的方式
  const stream = fs.createReadStream('The.Matrix.1080p.mkv');
  stream.pipe(gzip).pipe(res); // 大文件可以先压缩在转换为流传输
});

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24

// 客户端示例 XMLHttpRequest
var xhr = new XMLHttpRequest();
var received = 0;
var result = '';
xhr.open('post', '/httpstream', true);
xhr.setRequestHeader('Content-Type', 'application/json');
xhr.onreadystatechange = function () {
  if (xhr.readyState === 3) {
    //readystate 3 表示正在解析数据
    result = xhr.responseText.substring(received); //截取最新的数据
    received += result.length;
    console.log(result);
  }
};
const json = JSON.stringify({ name: 'test1' }); // post 请求数据，若为空可不传
xhr.send(json);

// 客户端示例 axios
const fnTest = (data: any, onDownloadProgress: any) => {
  return axios.request({
    method: 'post',
    url: '/httpstream',
    data,
    onDownloadProgress,
  });
};

async function doTest() {
  const params = {};
  let offset = 0;
  await fnTest(params, ({ target: xhr }) => {
    const { responseText } = xhr;
    const chunk = responseText.substring(offset);
    // 记录这一次的结束位置
    offset = responseText.length;
    console.log(chunk);
  });
}

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38

随着不断从服务器端接收到数据，客户端的 readyState 会周期性的变成 3，responseText 包含所有的数据源。通过 received 来记录之前已经处理过的数据长度，然后在 responseText 中截取最新的数据。

服务端部署：

# 仅展示有关配置
location /data {
        proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
        proxy_set_header X-Forwarded-Proto $scheme;
        proxy_set_header X-Real-IP $remote_addr;
        proxy_set_header Host $http_host;
        proxy_pass http://localhost:3100;
        # 关闭代理缓冲区，所有响应实时输出
        proxy_buffering off;
}

1
2
3
4
5
6
7
8
9
10

优点：

页面的整个生命周期内，只需要建立一个 http 连接

缺点：

如果接入的客户端过多，服务器端会因为 http 连接有限而无法为新的客户端提供服务
客户端接收到的数据流会越来越大，最终可能会引发页面的性能问题

# 4、SSE

SSE(Server-Sent Events)是基于 HTTP 实现的一套服务器向客户端发送数据的 API。他是针对上面说到的三种方法（轮询，长轮询，HTTP 流）的一个标准 API 实现。

使用 SSE API 可以创建到服务器端的但相连接，服务器可以通过这个连接发送任意数据。它有以下特点：

断开自动连接
服务器响应的 MIME 类型必须是 text/event-stream
需要浏览器 API 支持(参考浏览器兼容性 (opens new window))

使用方法如下：

//客户端js
var source = new EventSource(url);
//建立连接时触发
source.onopen = function () {
  //do something here
};
//从服务器端接收到新的事件时触发
source.onmessage = function (event) {
  var data = event.data; //服务器返回的数据存放在event.data中
};
//连接异常时触发
source.onerror = function () {
  //do something here
};

1
2
3
4
5
6
7
8
9
10
11
12
13
14

客户端创建一个 EventSource 对象，绑定到对应的 url，然后监听该对象的 onmessage 事件就可以获取到最新的数据。

// server端示例(nodejs)
let express = require('express');
let app = express();
app.use(express.static('resources'));
app.get('/httpstream', function (req, res) {
  var x = 0;
  res.writeHead(200, {
    'Content-Type': 'text/event-stream',
    'Cache-Control': 'no-cache',
    Connection: 'keep-alive',
  });
  //每个1s往客户端发送一条数据
  setInterval(function () {
    res.write('data: ' + x++ + '\n\n'); //发送的数据格式必须是"data: <内容>/n/n"
  }, 1000);
});
app.listen(3000);