分类
js

用 express.js 实现流式输出 HTTP 响应

0. 前言

我们先来总结一下客户端与服务器端的数据交互方式:

  1. ajax,即 XMLHttpRequest,最常见的数据交互方式,适用于较轻量级的一次性请求,比如用户登录、CRUD 等。
  2. SSE,服务器端事件,适用于有大量服务器端推送、且不需要从客户端发送数据的需求,比如盯股票行情
  3. WebSocket,适用于上下行数据都很多、都很频繁的情况,比如在线聊天、网络游戏等

单从技术角度来看,大概是这几个。根据需求变化,我也可以组合出更多方案。比如,类似 Showman 生成视频这种需求,耗时很长,对服务器资源占用很高,就比较适合:

  1. 发起请求,产生一条任务记录,进入队列
  2. 服务器上运行独立的进程,从队列中取任务出来执行,并记录日志
  3. 客户端不断检查任务进度

1. 需求

今天的需求,介于轻量一次性,与需要服务器长时间执行之间,即 FastTest 中的发布功能。这里,我们有以下需求:

  1. 管理员点击 Publish 按钮,开始发布静态网站
  2. 服务器需要先保存数据,然后调用 webpack 发布所有语言版本
  3. 只有少数管理员会使用此功能,服务器压力不大
  4. 管理员希望能了解发布进度,及每一步的状态

所以,提交任务后等待服务器慢慢跑就不合适;等待请求完成一次性看到所有结果也不合适;最合适的,就是基于长链接,不断获取返回信息,并在前端实时看到进度。也即是:流式输出 HTTP 响应。

2. 实现

2.1 后端

后端实现我选择 node.js + express.js,在后端领域,这两个我比较熟悉。express.js 是在 node.js 网络模块上进行封装得来,使用起来很简单,也支持原生 node.js 方法。

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

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

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

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

app.post('/data', async(req, res, next) => {
  res.setHeader('Content-type', 'application/octet-stream');
  
  // 第一次返回
  res.write('Local data saved. Start to build dist。files.\n');

  // 数次返回
  for (const item of items) {
    await doSomething(item);
    res.write(`${item} done successfully.\n`);
  }

  // 最后,全部完成
  res.write('All done.');
  // 关闭输出流
  res.end();
});

2.2 axios

axios 是很流行的 ajax 库,它进行了 Promise 封装,用起来很方便。这里我们要用 onDownloadProgress(即 axios 对 XMLHttpRequest.progress 的封装)获取下载进度,它会在每次服务器返回响应字符串的时候更新,我们只需要截取上次响应之后,这次响应新增的内容,即可。

function publish(data, onDownloadProgress) {
  return axios.post('/data', data, {
    onDownloadProgress,
  });
}

async function doPublish() {
  if (isLoading.value) {
    return;
  }

  isPublishing.value = true;
  message.value = status.value = null;

  try {
    const { cases, lang } = store.state;
    let offset = 0;
    await publish({ cases, lang }, ({ target: xhr }) => {
      // responseText 包含了从一开始到此刻的全部响应内容,所以我们需要从上次结束的位置截取,获得新增的内容
      const { responseText } = xhr;
      const chunk = responseText.substring(offset);
      // 记录这一次的结束位置
      offset = responseText.length;
      currentStatus.value = chunk;
    });
    status.value = true;
    currentStatus.value = '';
    message.value = 'Published successfully.';
  } catch (e) {
    message.value = 'Failed to publish. ' + e.message;
  }
  isPublishing.value = false;
}

2.3 Vue3

相对来说,Vue3 的部分最容易。这里我用了 animate.css 的 flash 效果,让信息更显眼。除此之外,就是简单的赋值。

<template lang="pug">
.alert.alert-info.mb-0.me-2.py-1.px-3.animated.flash.infinite.slower(
  v-if="currentStatus",
) {{currentStatus}}
</template>

3. 效果演示

流式输出效果演示

4. 部署

一般来说,我们很少会直接用 node.js 当服务器,多半会启动 node.js 服务,然后用 nginx 反向代理给用户访问。这里需要注意,nginx 默认会将响应内容存入缓冲区,然后批量返回给客户端。这会导致流式输出无效,变成常规的执行完毕后一次性输出。

所以我们必须修改 nginx 的配置:

# 仅展示有关配置
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;
}

5. 总结 & 扩展阅读

前些天我帮朋友做了个小项目,叫 FastTest。项目很简单,不过我主动扩展了技术选型,使用了非常多的技术,都是浅显入门的程度,非常适合用来做基础入门学习。本文也是从这个项目中提炼出来的。建议大家有空的时候看看,拉到本地,跑起来试试看效果。

文中的几段代码,分别位于:

项目的详细介绍请见:超全面全栈入门项目:fastest

扩展阅读