Express启用Gzip压缩后响应数据丢失问题排查

问题出在你把 _read() 写成异步函数了——Node.js 可读流的 _read() 设计逻辑不支持直接用 async/await，这会打乱流的背压机制，导致数据丢失或提前结束。

为什么异步_read()会出问题？ #

Node.js 可读流的 _read() 是同步触发、异步完成的接口，但不能声明为 async 函数。一旦用 async/await 包裹，_read() 会立即返回一个 Promise，流的内部逻辑会误以为你已经完成了数据读取，可能在Redis异步查询还没返回时就提前关闭流，或者因为背压处理混乱，导致大量数据没来得及推送就被截断。这就是你启用gzip后数据量不稳定的核心原因。

正确实现Redis游标流的方式 #

把 _read() 改回同步声明，内部用回调处理Redis的异步操作，同时手动控制读取状态避免并发调用。以下是修改后的 RedisQueryStream 示例：

const { Readable } = require('node:stream');
const RedisController = require('./RedisController');

class RedisQueryStream extends Readable {
  constructor(options = {}) {
    super({ ...options, objectMode: true }); // 开启objectMode，直接传输对象
    this.redisCtrl = options.redisController || new RedisController();
    this.cursor = 0;
    this.isFetching = false; // 防止重复触发异步查询
  }

  _read() {
    // 如果正在查询Redis，直接返回，避免并发
    if (this.isFetching) return;
    this.isFetching = true;

    // 调用Redis SCAN命令分批拿数据
    this.redisCtrl.scan(this.cursor, (err, [newCursor, keys]) => {
      if (err) {
        this.isFetching = false;
        this.emit('error', err);
        return;
      }

      this.cursor = parseInt(newCursor, 10);

      // 批量查询key对应的value
      this.redisCtrl.mget(keys, (err, values) => {
        if (err) {
          this.isFetching = false;
          this.emit('error', err);
          return;
        }

        // 逐个把解析后的对象推送到流里
        for (const val of values) {
          try {
            const data = JSON.parse(val);
            // push返回false时，说明流缓冲区满了，暂停读取，等drain事件
            if (!this.push(data)) {
              this.isFetching = false;
              return;
            }
          } catch (parseErr) {
            this.emit('error', parseErr);
          }
        }

        // 游标为0，说明所有数据读完了，结束流
        if (this.cursor === 0) {
          this.isFetching = false;
          this.push(null); // 标记流结束
          return;
        }

        // 缓冲区还有空间，继续读下一批
        this.isFetching = false;
        this._read();
      });
    });

    // 监听drain事件，缓冲区空了之后继续读取
    this.once('drain', () => {
      if (!this.isFetching && this.cursor !== 0) {
        this._read();
      }
    });
  }
}

module.exports = RedisQueryStream;

配合Express和Gzip的正确用法 #

用 stream.pipeline 串联流，同时设置正确的响应头，确保客户端能解析压缩数据：

const express = require('express');
const { pipeline, Transform } = require('node:stream');
const zlib = require('node:zlib');
const RedisQueryStream = require('./RedisQueryStream');

const app = express();

app.get('/stream-data', (req, res) => {
  // 告诉客户端数据是gzip压缩的
  res.setHeader('Content-Encoding', 'gzip');
  res.setHeader('Content-Type', 'application/json; charset=utf-8');

  const redisStream = new RedisQueryStream();
  const jsonTransform = new Transform({
    writableObjectMode: true,
    transform(obj, _, cb) {
      // 把对象转成JSON字符串，加换行符方便客户端逐行处理
      cb(null, JSON.stringify(obj) + '\n');
    }
  });
  const gzip = zlib.createGzip();

  // 用pipeline自动处理流的错误、关闭和清理
  pipeline(redisStream, jsonTransform, gzip, res, (err) => {
    if (err) {
      console.error('流处理失败:', err);
      if (!res.headersSent) res.status(500).send('Server Error');
    }
  });
});

app.listen(3000, () => console.log('Server running on port 3000'));

几个关键要点 #

1. 绝对不能把_read()写成async函数：必须用回调控制异步流程，保证流的背压机制正常工作。
2. 处理背压：当push()返回false时，要暂停读取，等drain事件触发后再继续。
3. 标记流结束：Redis游标为0时，一定要调用push(null)来结束可读流，否则下游会一直等待数据。
4. 设置正确响应头：启用gzip必须加Content-Encoding: gzip，否则客户端会把压缩数据当成普通JSON解析，导致错误。
5. 用pipeline管理流：它会自动处理所有流的生命周期，避免内存泄漏或数据丢失。

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