Node.js 中级实战：基于 Worker Threads 与流式处理构建高并发文件处理服务

在 Node.js 里做文件处理，很多人第一反应是：fs.createReadStream() 已经很高效了，为什么还要折腾 Worker Threads？

我一开始也这么想。直到服务一上量：上传文件多了、压缩和哈希任务堆起来、日志里开始出现事件循环延迟变大、接口响应时间抖动，才真正意识到——I/O 高效，不等于整体服务就能扛住高并发。
尤其是“文件读取 + CPU 计算 + 结果落盘/回传”这类混合型任务，如果只靠主线程，很容易把 Node.js 的单线程优势变成瓶颈。

这篇文章我们从架构角度，做一个中级实战：用流式处理解决内存问题，用 Worker Threads 解决 CPU 问题，再加上任务编排、限流和错误治理，构建一个可运行的高并发文件处理服务。

背景与问题

先看一个典型需求：

用户上传大文件
服务端需要对文件做：
- 分块读取
- 内容转换
- 哈希计算 / 文本分析 / 压缩等 CPU 密集型处理
- 输出到结果文件
同时系统要支持较高并发

很多项目早期会写成这样：

主线程接收请求
主线程读取整个文件
主线程做计算
主线程一次性写出结果

这个方案在小文件、低并发时没问题，但一上量就会暴露几个老问题：

1. 内存占用高

如果用 fs.readFile() 或把大块数据长期堆在内存里，几十个并发大文件就能把进程吃满。

2. 事件循环被 CPU 任务拖慢

Node.js 主线程擅长协调 I/O，不擅长长时间跑重计算。
比如做全文扫描、压缩、加解密、复杂解析，都会影响其他请求的响应。

3. 缺少背压控制

上游读得太快，下游写得太慢，中间缓存会不断膨胀。
这时候“能跑”和“能稳跑”是两回事。

4. 异常传播混乱

多线程、流式管道、文件 I/O 混在一起时，如果错误处理没设计好，结果往往是：

文件句柄没关
临时文件残留
Worker 未回收
请求超时但后台还在跑

方案目标

我们要的不是“能处理文件”，而是一个更工程化的目标：

大文件不一次性读入内存
CPU 密集任务从主线程剥离
并发可控，不把机器打满
支持背压，避免流量洪峰时内存抖动
错误可观测、可回收、可降级

这也是本文的架构主线。

方案总览

我们把整个文件处理链路拆成三层：

接入层：HTTP 服务接收任务请求
调度层：任务队列 + Worker 池，限制并发
处理层：流式读取文件，按块交给 Worker 处理，再流式写出结果

flowchart LR
    A[客户端请求] --> B[Node.js HTTP 服务]
    B --> C[任务调度器]
    C --> D[Worker 池]
    B --> E[ReadStream]
    E --> F[分块派发]
    F --> D
    D --> G[结果聚合]
    G --> H[WriteStream]
    H --> I[输出文件]

这个设计的关键点是：

流负责控制内存和背压
Worker Threads 负责吃掉 CPU 密集任务
调度器 负责限制并发和复用线程，避免频繁创建销毁 Worker

核心原理

这一节不只是讲 API，而是讲清楚为什么这种组合有效。

一、流式处理：解决“大文件 + 高并发”的内存问题

Node.js 的 Stream 天生适合文件场景：

读取端按块吐数据
处理端按块消费
写入端按块落盘
下游慢时，上游自动减速

也就是所谓的背压（backpressure）。

如果你直接 readFile，本质是“先把全部数据拿到手再说”；
如果你用 Stream，本质是“来一块，处理一块，写一块”。

对于高并发文件服务，这个差异非常关键。

二、Worker Threads：解决 CPU 密集任务阻塞主线程的问题

Node.js 主线程本质上是一个事件循环。
它非常适合：

网络请求
文件 I/O 协调
定时器
数据分发

但不适合长时间做下面这些事：

大量文本分析
压缩
加密/解密
图像处理
哈希批量计算
自定义规则扫描

Worker Threads 允许你在同一个 Node.js 进程里开多个线程，主线程把任务派给 Worker 执行，自己继续处理其他请求。

注意一个常见误区：
Worker Threads 不是为了替代 Stream，而是和 Stream 分工合作。

Stream：解决“数据怎么稳稳地流动”
Worker：解决“谁来做重计算”

三、为什么要用 Worker 池，而不是每个任务新建一个 Worker

理论上你可以每次来一个文件块，就 new Worker()。
但实际上这会非常浪费：

线程创建有成本
消息通信有成本
频繁销毁会导致抖动
高并发下线程数不可控

更合理的方式是：

预先创建固定数量的 Worker
做成一个简单线程池
按需把任务派给空闲 Worker
忙不过来时排队

这就是“高并发服务”和“脚本程序”的分水岭。

四、消息传递与数据复制成本

主线程和 Worker 之间通过 postMessage() 通信。
这里有一个性能重点：大 Buffer 频繁复制会带来额外开销。

在 Node.js 里可以通过 transfer list 转移底层 ArrayBuffer，减少复制成本。
但也要注意：

转移后原对象在发送侧不可再用
如果逻辑里还要复用原 Buffer，要谨慎设计

我当时第一次做时，没注意 transfer 之后 Buffer 失效，结果下游写文件时直接拿到空数据，查了半天。

架构设计与取舍分析

方案 A：主线程全处理

优点：

简单
开发快

缺点：

CPU 任务阻塞事件循环
并发上来就抖
大文件场景风险大

适用：

小工具、低并发后台任务

方案 B：仅用 Stream，不用 Worker

优点：

内存控制好
代码比多线程简单

缺点：

CPU 计算仍在主线程
遇到复杂处理时吞吐有限

适用：

主要是 I/O 搬运、轻量转换

方案 C：Stream + Worker 池

优点：

内存可控
CPU 与 I/O 分离
并发更稳
可扩展性强

缺点：

架构复杂度更高
需要处理任务调度、顺序、错误回收

适用：

中高并发文件处理服务
既有大文件，又有较重计算

这也是本文最终采用的方案。

数据流与线程协作时序

sequenceDiagram
    participant Client
    participant Main as 主线程HTTP服务
    participant Stream as 文件读取流
    participant Pool as Worker池
    participant Worker as Worker线程
    participant Writer as 写入流

    Client->>Main: 提交文件处理请求
    Main->>Stream: 创建 ReadStream
    loop 按块读取
        Stream->>Main: chunk
        Main->>Pool: 派发任务
        Pool->>Worker: postMessage(chunk)
        Worker-->>Pool: 返回处理结果
        Pool-->>Main: result
        Main->>Writer: 写入处理后的chunk
    end
    Writer-->>Main: finish
    Main-->>Client: 返回成功

容量估算：线程数、块大小、并发数怎么选

这个问题没有万能值，但可以按经验做初始配置。

1. Worker 数量

通常建议从 CPU 核心数 - 1 或 CPU 核心数 开始试。

例如 8 核机器：

Worker 池先设 6~8
主线程留资源处理网络和调度

如果任务非常重，线程太多反而会产生抢占和上下文切换开销。

2. 文件块大小

常见起点：

64KB
256KB
1MB

经验上：

块太小：消息传递次数过多，调度开销上升
块太大：单次内存峰值变高，处理延迟增大

对于“文本扫描 + 转换”这种任务，我一般先从 256KB 开始压测。

3. 请求级并发

就算有 Worker 池，也不要让无限多请求同时挤进来。
建议设置：

单实例最大并发任务数
超限后排队或返回 429/503
配合进程管理器或容器水平扩容

实战代码（可运行）

下面我们做一个可运行的示例服务：

输入：一个本地文件路径
处理：按块读取，把内容交给 Worker 做“模拟 CPU 密集转换”
输出：写入新文件
特点：
- 使用 Worker 池
- 支持并发限制
- 使用流式处理
- 按顺序写回结果

示例基于 Node.js 18+。

目录结构

file-service/
├─ package.json
├─ server.js
├─ worker-pool.js
└─ file-worker.js

package.json

{
  "name": "file-service",
  "version": "1.0.0",
  "type": "commonjs",
  "scripts": {
    "start": "node server.js"
  }
}

file-worker.js

这个 Worker 做一件事：接收一个 chunk，执行一个相对耗 CPU 的处理逻辑。
为了便于演示，我们做一个“字符翻转 + 简单累积计算”的模拟任务。

const { parentPort } = require('worker_threads');

function cpuHeavyTransform(buffer) {
  const text = buffer.toString('utf8');

  // 模拟较重 CPU 计算
  let score = 0;
  for (let i = 0; i < text.length; i++) {
    score += text.charCodeAt(i) % 97;
  }

  // 做一个可见的文本变换
  const transformed = text.toUpperCase().split('').reverse().join('');

  return Buffer.from(`[score=${score}]` + transformed, 'utf8');
}

parentPort.on('message', (message) => {
  const { taskId, chunk } = message;

  try {
    const result = cpuHeavyTransform(Buffer.from(chunk));
    parentPort.postMessage({ taskId, result });
  } catch (error) {
    parentPort.postMessage({
      taskId,
      error: {
        message: error.message,
        stack: error.stack
      }
    });
  }
});

worker-pool.js

这里实现一个简单的 Worker 池：

固定数量 Worker
空闲就执行任务
忙时排队
任务以 Promise 形式返回

const { Worker } = require('worker_threads');
const os = require('os');
const path = require('path');

class WorkerPool {
  constructor(options = {}) {
    this.size = options.size || Math.max(1, os.cpus().length - 1);
    this.workerFile = options.workerFile || path.resolve(__dirname, 'file-worker.js');

    this.workers = [];
    this.idleWorkers = [];
    this.taskQueue = [];
    this.callbacks = new Map();
    this.taskId = 0;

    for (let i = 0; i < this.size; i++) {
      this._createWorker();
    }
  }

  _createWorker() {
    const worker = new Worker(this.workerFile);

    worker.on('message', (message) => {
      const { taskId, result, error } = message;
      const callback = this.callbacks.get(taskId);

      if (!callback) return;

      this.callbacks.delete(taskId);

      if (error) {
        callback.reject(new Error(error.message));
      } else {
        callback.resolve(Buffer.from(result));
      }

      this.idleWorkers.push(worker);
      this._drainQueue();
    });

    worker.on('error', (err) => {
      console.error('[worker error]', err);
    });

    worker.on('exit', (code) => {
      this.workers = this.workers.filter((w) => w !== worker);
      this.idleWorkers = this.idleWorkers.filter((w) => w !== worker);

      if (code !== 0) {
        console.error(`[worker exit] code=${code}, recreating...`);
        this._createWorker();
      }
    });

    this.workers.push(worker);
    this.idleWorkers.push(worker);
  }

  exec(chunk) {
    return new Promise((resolve, reject) => {
      const taskId = ++this.taskId;
      this.taskQueue.push({ taskId, chunk, resolve, reject });
      this._drainQueue();
    });
  }

  _drainQueue() {
    while (this.idleWorkers.length > 0 && this.taskQueue.length > 0) {
      const worker = this.idleWorkers.shift();
      const task = this.taskQueue.shift();

      this.callbacks.set(task.taskId, {
        resolve: task.resolve,
        reject: task.reject
      });

      worker.postMessage({
        taskId: task.taskId,
        chunk: task.chunk
      });
    }
  }

  async destroy() {
    await Promise.all(this.workers.map((worker) => worker.terminate()));
  }
}

module.exports = WorkerPool;

server.js

这里用原生 http 模块做一个简单 HTTP 服务，避免引入额外依赖。
接口形式：

POST /process
Content-Type: application/json

{
  "input": "./input.txt",
  "output": "./output.txt"
}

核心逻辑：

主线程通过 createReadStream 分块读取
每个 chunk 提交给 Worker 池
为了保证输出顺序，按序号缓存结果后再写入
超过最大任务数时拒绝新请求

const http = require('http');
const fs = require('fs');
const path = require('path');
const { once } = require('events');
const WorkerPool = require('./worker-pool');

const pool = new WorkerPool({
  size: 4,
  workerFile: path.resolve(__dirname, 'file-worker.js')
});

const MAX_CONCURRENT_JOBS = 4;
let activeJobs = 0;

async function processFile(inputPath, outputPath) {
  return new Promise((resolve, reject) => {
    const readStream = fs.createReadStream(inputPath, {
      highWaterMark: 256 * 1024
    });

    const writeStream = fs.createWriteStream(outputPath);

    let index = 0;
    let nextWriteIndex = 0;
    const pendingResults = new Map();
    const runningTasks = new Set();
    let streamEnded = false;
    let failed = false;

    function cleanup(err) {
      if (failed) return;
      failed = true;
      readStream.destroy();
      writeStream.destroy();
      reject(err);
    }

    async function tryFlush() {
      while (pendingResults.has(nextWriteIndex)) {
        const buffer = pendingResults.get(nextWriteIndex);
        pendingResults.delete(nextWriteIndex);

        const canContinue = writeStream.write(buffer);
        nextWriteIndex++;

        if (!canContinue) {
          await once(writeStream, 'drain');
        }
      }

      if (streamEnded && runningTasks.size === 0 && pendingResults.size === 0) {
        writeStream.end();
      }
    }

    readStream.on('data', async (chunk) => {
      readStream.pause();

      const currentIndex = index++;
      const taskPromise = pool.exec(chunk)
        .then(async (result) => {
          runningTasks.delete(taskPromise);
          pendingResults.set(currentIndex, result);
          await tryFlush();

          if (!failed) {
            readStream.resume();
          }
        })
        .catch((err) => {
          runningTasks.delete(taskPromise);
          cleanup(err);
        });

      runningTasks.add(taskPromise);
    });

    readStream.on('end', async () => {
      streamEnded = true;
      try {
        await Promise.all([...runningTasks]);
        await tryFlush();
      } catch (err) {
        cleanup(err);
      }
    });

    readStream.on('error', cleanup);
    writeStream.on('error', cleanup);
    writeStream.on('finish', resolve);
  });
}

function sendJson(res, statusCode, data) {
  res.writeHead(statusCode, {
    'Content-Type': 'application/json; charset=utf-8'
  });
  res.end(JSON.stringify(data));
}

const server = http.createServer((req, res) => {
  if (req.method === 'POST' && req.url === '/process') {
    if (activeJobs >= MAX_CONCURRENT_JOBS) {
      return sendJson(res, 429, {
        error: 'Too many concurrent jobs'
      });
    }

    let raw = '';
    req.on('data', (chunk) => {
      raw += chunk;
      if (raw.length > 1024 * 1024) {
        req.destroy(new Error('Request body too large'));
      }
    });

    req.on('end', async () => {
      try {
        const body = JSON.parse(raw || '{}');
        const input = path.resolve(body.input);
        const output = path.resolve(body.output);

        if (!fs.existsSync(input)) {
          return sendJson(res, 400, { error: 'Input file does not exist' });
        }

        activeJobs++;
        const start = Date.now();

        await processFile(input, output);

        activeJobs--;
        return sendJson(res, 200, {
          message: 'File processed successfully',
          durationMs: Date.now() - start,
          output
        });
      } catch (err) {
        activeJobs = Math.max(0, activeJobs - 1);
        return sendJson(res, 500, {
          error: err.message
        });
      }
    });

    req.on('error', (err) => {
      return sendJson(res, 500, { error: err.message });
    });

    return;
  }

  sendJson(res, 404, { error: 'Not found' });
});

server.listen(3000, () => {
  console.log('Server is running at http://localhost:3000');
});

process.on('SIGINT', async () => {
  console.log('\nShutting down...');
  await pool.destroy();
  process.exit(0);
});

如何运行

1. 准备测试文件

echo "hello node worker threads and stream" > input.txt

或者准备更大的文本文件做压测。

2. 启动服务

npm start

3. 发请求

curl -X POST http://localhost:3000/process \
  -H "Content-Type: application/json" \
  -d '{"input":"./input.txt","output":"./output.txt"}'

4. 查看输出

cat output.txt

处理链路状态图

这个状态图可以帮助你理解“请求生命周期”和“异常该在哪一层终止”。

stateDiagram-v2
    [*] --> 接收请求
    接收请求 --> 校验参数
    校验参数 --> 创建读写流
    创建读写流 --> 分块读取
    分块读取 --> 派发到Worker
    派发到Worker --> 等待结果
    等待结果 --> 顺序写入
    顺序写入 --> 分块读取: 还有数据
    顺序写入 --> 完成: 无剩余数据
    完成 --> [*]

    校验参数 --> 失败
    创建读写流 --> 失败
    分块读取 --> 失败
    派发到Worker --> 失败
    等待结果 --> 失败
    顺序写入 --> 失败
    失败 --> 清理资源
    清理资源 --> [*]

关键实现细节拆解

1. 为什么在 `data` 事件里 `pause()` / `resume()`

因为我们不希望读取速度无限快。
主线程拿到一个 chunk 后，需要等 Worker 处理完成、结果有机会写回后，再继续往下读。

这是一种比较直接的背压控制方式。

当然，更进一步你也可以封装成 Transform 流，但中级阶段先把调度关系看清楚更重要。

2. 为什么要给 chunk 编号

Worker 并发处理后，返回顺序不一定和读取顺序一致。
如果你直接“谁先回来谁先写”，输出文件内容就乱了。

所以我们做了两件事：

读取时给每块分配 index
写入时只按 nextWriteIndex 顺序刷出

这在“并发处理 + 顺序输出”的场景里是非常常见的技巧。

3. 为什么 `pendingResults` 不能无限大

如果 Worker 很快、写入很慢，结果缓存也可能堆积。
本文示例里主要依赖写流的 drain 和读取暂停来做基础控制。

生产环境里可以进一步限制：

最大在途任务数
最大待写结果数
每请求最大缓存字节数

否则极端情况下仍可能内存上涨。

常见坑与排查

这一节我尽量讲“真会踩到的坑”，而不是 API 手册里那种泛泛提醒。

坑 1：以为用了 Stream 就不会阻塞

现象：

内存不高，但接口还是慢
事件循环延迟明显
CPU 打满

原因：

你只是把读写做成了流式，但真正重的逻辑仍在主线程里跑。

排查方法：

用 top / htop 看 CPU
打印处理耗时
观察请求高峰时 Node 进程响应是否整体变慢

解决：

把 CPU 密集型逻辑迁移到 Worker 里。

坑 2：Worker 开太多，反而更慢

现象：

线程数增加了，但吞吐没涨
系统上下文切换增多
CPU 使用率很高但任务完成变慢

原因：

线程不是越多越好。
超过 CPU 核数太多后，会出现明显调度开销。

排查方法：

压测不同线程数：2、4、8、16
记录吞吐、P95、CPU 使用率

解决：

一般以 CPU 核数附近作为起点，压测后再调。

坑 3：输出文件内容错乱

现象：

结果文件内容顺序不对，尤其并发高时偶发。

原因：

多个 chunk 并发处理后，完成顺序乱了，但写入没做顺序控制。

解决：

给任务编号，按序写入。
这类问题在本地小文件测试很难暴露，压测时才明显。

坑 4：Worker 崩了，任务一直挂着

现象：

某些请求不返回，进程也没报明显错误。

原因：

Worker 异常退出后，未及时 reject 对应任务，或者线程池没补齐。

解决：

监听 error 和 exit
对 Worker 退出做自动重建
对在途任务增加超时机制

本文示例里做了基础重建，但严格来说还可以再补“任务超时清理”。

坑 5：背压没接住，内存还是涨

现象：

随着并发提升，内存持续走高。

原因：

虽然用了 Stream，但中间缓冲区、任务队列、待写结果没有上限。

解决：

限制请求级并发
限制池内队列长度
限制每个请求最大在途 chunk 数
控制 highWaterMark

坑 6：路径参数存在安全风险

现象：

接口允许用户传文件路径，可能被读写任意文件。

原因：

未做路径白名单或目录限制。

解决：

限制输入输出只能位于指定目录
做 path.resolve() 后校验前缀
禁止覆盖敏感文件

这个问题非常容易被忽略，但在文件服务里属于基本安全项。

常见排查清单

如果你的服务“能跑，但不稳”，我建议按下面顺序查：

看 CPU
- 是否主线程占满
- Worker 数是否过多
看内存
- 是否有结果缓存堆积
- 是否把大文件读进内存
看事件循环延迟
- 用 perf_hooks 监控 event loop delay
- 高延迟通常说明主线程被拖住
看队列长度
- Worker 池等待队列是否持续升高
- 请求级队列是否超限
看磁盘吞吐
- 有时瓶颈根本不在 Node，而在磁盘 I/O
看错误回收
- 异常时文件句柄、临时文件、Worker 是否及时释放

安全/性能最佳实践

这一节是最接近生产落地的部分。

安全实践

1. 限制文件路径作用域

不要让客户端直接指定任意绝对路径。
应将输入输出限制在预设目录：

function safeResolve(baseDir, targetPath) {
  const resolved = path.resolve(baseDir, targetPath);
  if (!resolved.startsWith(path.resolve(baseDir))) {
    throw new Error('Invalid path');
  }
  return resolved;
}

2. 限制请求体大小

无论是 JSON 参数还是上传内容，都要做大小限制。
否则很容易被大请求拖垮。

3. 避免覆盖关键文件

输出文件最好使用：

临时文件名
处理完成后原子替换
或使用唯一任务目录

4. 做任务超时

有些 Worker 任务可能进入异常状态，必须设置超时并中断回收。

性能实践

1. 优先复用 Worker 池

不要按 chunk 创建 Worker。
线程池几乎是生产环境必选项。

2. 选择合适的 `highWaterMark`

文件读块大小和任务粒度要一起调。
不是越大越快，要看：

CPU 处理模型
消息传输成本
磁盘吞吐

3. 限制在途任务数

可以在每个请求内设置例如“最多 8 个 chunk 在处理中”，避免结果积压。

4. 使用 Transferable 对象优化大块数据传递

当 chunk 很大、消息频繁时，可以考虑转移底层内存，减少复制：

worker.postMessage(
  { taskId, chunk: uint8Array },
  [uint8Array.buffer]
);

但前提是你明确知道发送后原对象不可继续使用。

5. 做分层指标监控

建议至少监控这些指标：

请求耗时
文件处理耗时
Worker 池队列长度
活跃 Worker 数
事件循环延迟
进程 RSS / heapUsed
读写错误数

没有指标，优化基本靠猜。

一个更贴近生产的增强建议

如果你准备把这个方案继续做大，可以往这几个方向演进：

1. HTTP 接入和任务执行解耦

对于超大文件或长耗时任务，不建议请求一直挂着等结果。
可以改成：

提交任务 -> 返回任务 ID
后台异步处理
轮询或回调获取结果

2. 引入持久化队列

单机内存队列适合入门和中等规模。
再往上走，可以考虑：

Redis 队列
RabbitMQ
Kafka（更偏流式平台）

3. 多进程 + 多线程混合

单个 Node 进程里有 Worker 池，但如果机器核数较多，还可以配合：

cluster
PM2
容器多副本

让接入层和处理层更均衡。

4. 针对不同任务类型分池

不要把“轻量文本处理”和“重压缩任务”混在一个 Worker 池里。
否则重任务容易把轻任务拖慢。

边界条件：什么时候不建议上 Worker Threads

虽然本文主推这个方案，但也不是所有场景都适合。

不建议使用的情况

1. 任务几乎全是 I/O，没有明显 CPU 计算

比如只是文件搬运、转存、上传对象存储。
这时候 Stream 就够了，多线程收益不大。

2. 任务特别短小

如果每个任务只处理几毫秒，线程通信成本可能比计算本身还高。

3. 已经有成熟的外部处理系统

例如图片、视频、压缩等任务已交给专用服务，Node 只负责编排。
那 Node 端保持轻量更合适。

总结

把这篇文章浓缩成一句话，就是：

在 Node.js 的高并发文件处理场景里，流式处理负责“稳”，Worker Threads 负责“快”，线程池和限流负责“可控”。

如果你正在做中级阶段的 Node.js 服务，我建议按下面这个顺序落地：

先把文件读取改成 Stream
- 解决大文件内存问题
识别 CPU 密集逻辑
- 把真正重的计算迁到 Worker
引入 Worker 池
- 固定线程数，避免频繁创建销毁
做好顺序控制与背压
- 给 chunk 编号
- 控制在途任务和待写缓存
补齐异常治理和监控
- 超时、重建、清理、指标

如果你只记一个经验值，我会建议你从这组初始配置开始压测：

Worker 数：CPU 核数 - 1
文件块：256KB
单实例并发任务：4~8
每请求在途 chunk：先限制在小范围

然后基于真实数据调，而不是凭感觉调。

高并发文件处理这件事，真正难的从来不是“写出功能”，而是让它在高负载下仍然稳定、可控、可定位。
而 Worker Threads + Stream，正是 Node.js 在这类场景里非常值得掌握的一组组合拳。

《Node.js 中级实战：基于 Worker Threads 与流式处理构建高并发文件处理服务》

Node.js 中级实战：基于 Worker Threads 与流式处理构建高并发文件处理服务

背景与问题

1. 内存占用高

2. 事件循环被 CPU 任务拖慢

3. 缺少背压控制

4. 异常传播混乱

方案目标

方案总览

核心原理

一、流式处理：解决“大文件 + 高并发”的内存问题

二、Worker Threads：解决 CPU 密集任务阻塞主线程的问题

三、为什么要用 Worker 池，而不是每个任务新建一个 Worker

四、消息传递与数据复制成本

架构设计与取舍分析

方案 A：主线程全处理

方案 B：仅用 Stream，不用 Worker

方案 C：Stream + Worker 池

数据流与线程协作时序

容量估算：线程数、块大小、并发数怎么选

1. Worker 数量

2. 文件块大小

3. 请求级并发

实战代码（可运行）

目录结构

package.json

file-worker.js

worker-pool.js

server.js

如何运行

1. 准备测试文件

2. 启动服务

3. 发请求

4. 查看输出

处理链路状态图

关键实现细节拆解

1. 为什么在 data 事件里 pause() / resume()

2. 为什么要给 chunk 编号

3. 为什么 pendingResults 不能无限大

常见坑与排查

坑 1：以为用了 Stream 就不会阻塞

坑 2：Worker 开太多，反而更慢

坑 3：输出文件内容错乱

坑 4：Worker 崩了，任务一直挂着

坑 5：背压没接住，内存还是涨

坑 6：路径参数存在安全风险

常见排查清单

安全/性能最佳实践

安全实践

1. 限制文件路径作用域

2. 限制请求体大小

3. 避免覆盖关键文件

4. 做任务超时

性能实践

1. 优先复用 Worker 池

2. 选择合适的 highWaterMark

3. 限制在途任务数

4. 使用 Transferable 对象优化大块数据传递

5. 做分层指标监控

一个更贴近生产的增强建议

1. HTTP 接入和任务执行解耦

2. 引入持久化队列

3. 多进程 + 多线程混合

4. 针对不同任务类型分池

边界条件：什么时候不建议上 Worker Threads

不建议使用的情况

1. 任务几乎全是 I/O，没有明显 CPU 计算

2. 任务特别短小

3. 已经有成熟的外部处理系统

总结

1. 为什么在 `data` 事件里 `pause()` / `resume()`

3. 为什么 `pendingResults` 不能无限大

2. 选择合适的 `highWaterMark`