如何在Express与Angular间实现HTTP流式响应传输

当前代码两端均不符合流式分段传输的实现要求，核心阻塞点在Angular端HTTP客户端的默认行为，Express端也存在缓冲未处理的问题：

1. Express端问题 #

• 仅设置Transfer-Encoding: chunked请求头无法保证分段内容立即下发：Node.js原生HTTP层、Express常用的compression压缩中间件、部署时前置的Nginx等反向代理，默认都会对小体积响应做缓冲，攒到固定大小或响应结束才会真正推送给客户端。你每次写入的内容长度过短，会直接被缓冲层拦截。
• 每次写入分段内容后没有主动触发缓冲刷新，内容会一直留在缓冲区内直到请求结束。

2. Angular端核心问题 #

• 你使用的HttpClient.get()默认行为是等待整个HTTP响应完全接收完成，才会把完整响应体传给subscribe回调，无论后端是否分块发送，这个API本身不会在收到中间数据块时触发事件，所以永远只能等所有内容传输完成后拿到一次全量数据。
• 不要尝试用HttpClient的reportProgress选项实现该需求，它只会返回已加载的字节数，不会暴露已接收的具体响应内容，无法读取分段数据。

Express端修正代码 #

每次写入分段后主动刷新缓冲，同时添加禁用代理缓冲的请求头：

router.get("/getStreamedData", (request, response, next) => {
  // 基础响应头设置，不要用text/html，避免浏览器预解析缓冲
  response.setHeader('Content-Type', 'text/plain; charset=utf-8');
  response.setHeader('Transfer-Encoding', 'chunked');
  // 禁用Nginx等反向代理缓冲
  response.setHeader('X-Accel-Buffering', 'no');
  response.setHeader('Cache-Control', 'no-cache');

  // 兼容不同场景的缓冲刷新方法
  const flush = () => {
    if (typeof response.flush === 'function') {
      // 适配compression中间件的flush方法
      response.flush();
    } else if (response.socket?.flush) {
      // 原生Node.js TCP层刷缓冲
      response.socket.flush();
    }
  };

  console.log("Partial response 1");
  response.write("Partial answer 1\n"); // 加换行作为分段分隔符
  flush(); // 写完立刻刷新缓冲推送

  setTimeout(function () {
    console.log("Partial response 2");
    response.write("Partial answer 2\n");
    flush();

    setTimeout(function () {
      console.log("Partial response 3");
      response.write("Partial answer 3\n");
      flush();          
      setTimeout(function () {
        console.log("response end");
        response.end()  
      }, 5000)  
    }, 5000)
  }, 5000)
});

Angular端修正代码 #

放弃默认HttpClient的全量接收逻辑，改用原生fetch API实现流式读取，服务层改造为返回分段数据的可观察对象：

// 服务层代码
import { Observable, Subject } from 'rxjs';
import { HttpParams } from '@angular/common/http';

getProduktdataDatastore(language: string): Observable<string> {
  const chunkSubject = new Subject<string>();
  const params = new HttpParams().set('language', language);
  const requestUrl = `${this.apiPath}getStreamedData?${params.toString()}`;

  // 发起流式请求
  fetch(requestUrl, { method: 'GET' })
    .then(async (response) => {
      if (!response.ok) throw new Error(`请求失败，状态码：${response.status}`);
      const reader = response.body!.getReader();
      const decoder = new TextDecoder('utf-8');
      let pendingText = '';

      // 循环读取每一个到达的数据块
      while (true) {
        const { done, value } = await reader.read();
        if (done) break;
        // 解码二进制块为文本
        const chunkStr = decoder.decode(value, { stream: true });
        pendingText += chunkStr;
        // 按换行分割完整分段，未接收完成的分段留到下次拼接
        const fullChunks = pendingText.split('\n');
        pendingText = fullChunks.pop() || '';
        // 把完整分段推送给订阅者
        fullChunks.forEach(chunk => chunk.trim() && chunkSubject.next(chunk.trim()));
      }
      // 推送最后一段剩余内容
      if (pendingText.trim()) chunkSubject.next(pendingText.trim());
      chunkSubject.complete();
    })
    .catch(err => chunkSubject.error(err));

  return chunkSubject.asObservable();
}

组件层订阅逻辑无需大改，现在每收到一个分段就会触发一次next回调，可实时处理：

// 组件层代码
this.WebGridService.getProduktdataDatastore(this.selectedLanguage)                   
  .subscribe(
    singleChunk => {
      // 每收到一个分段立刻触发，无需等待全量返回
      console.log("实时收到分段数据: ", singleChunk);
      // 在此处编写实时渲染/处理逻辑
    },
    error => {
      this.flagLoadData = false; 
      this.messageBox =  {messageBoxFlag: true,  text: error.message}  ;
    },
    () => {
      // 所有分段传输完成的回调
      this.flagLoadData = false;
      console.log("流式传输完成");
    }
  )

• 如果需要传输结构化数据，直接使用标准SSE（Server-Sent Events）协议即可，Express端可以直接用res.writeEvent等现成方法，Angular端用原生EventSource对接，不需要手动处理流分割、缓冲问题，稳定性更高。
• 如果服务前部署了Nginx，除了添加X-Accel-Buffering: no头，还要确认配置中没有强制开启proxy_buffering on，否则代理层还是会攒住响应。

内容的提问来源于stack exchange，提问作者High Malloy