Node.js异步任务协作实战：7种Promise模式与并发控制方案

脚本专家

在 Node.js 开发中，经常需要同时处理多个独立的异步任务，例如读取配置文件、调用外部接口、批量上传文件等。这些任务彼此独立，但最终结果需要协同处理。如果逐个 await，性能会大打折扣；如果盲目并发，又可能引发资源耗尽或错误处理混乱。本文基于真实项目场景，介绍 7 种成熟的异步协作方案，每种都配有可直接运行的代码示例。

1. Promise.all——应用启动加载必要配置
场景：服务启动时必须读取数据库、Redis 和第三方密钥三个配置文件，任何一个缺失或格式错误都不能继续启动。

2. const fs = require('fs').promises;
3. async function loadConfigs() {
4.   const [db, redis, secrets] = await Promise.all([
5.     fs.readFile('./config/db.json', 'utf8').then(JSON.parse),
6.     fs.readFile('./config/redis.json', 'utf8').then(JSON.parse),
7.     fs.readFile('./config/secrets.json', 'utf8').then(JSON.parse)
8.   ]);
9.   console.log('所有配置加载完成', { db, redis, secrets });
10. }

复制代码

特点：全成功或全失败，结果以数组顺序返回。适合“缺一不可”的场景。

2. Promise.allSettled——批量同步用户数据到多个外部系统
场景：用户更新个人资料后，需要同步到 CRM、邮件服务、推送系统。允许个别失败，但要记录失败原因，后续重试。

2. async function syncUserToExternal(user) {
3.   const tasks = [
4.     syncToCRM(user),
5.     syncToEmailService(user),
6.     syncToPushService(user)
7.   ];
8.   const results = await Promise.allSettled(tasks);
9.   const failed = results.filter(r => r.status === 'rejected');
10.   if (failed.length) {
11.     console.error(`同步失败 ${failed.length} 个系统`, failed.map(f => f.reason));
12.     // 将失败记录到数据库，等待重试队列处理
13.   }
14.   return results;
15. }

复制代码

特点：等待所有任务完成，无论成功或失败，都能拿到每个任务的最终状态。

3. Promise.race——HTTP 请求超时控制
场景：调用外部 API，必须在 3 秒内返回结果，否则自动降级使用缓存数据。

2. function fetchWithTimeout(url, timeout = 3000) {
3.   const controller = new AbortController();
4.   const fetchPromise = fetch(url, { signal: controller.signal });
5.   const timeoutPromise = new Promise((_, reject) =>
6.     setTimeout(() => {
7.       controller.abort();
8.       reject(new Error('请求超时'));
9.     }, timeout)
10.   );
11.   return Promise.race([fetchPromise, timeoutPromise]);
12. }
13. try {
14.   const data = await fetchWithTimeout('https://slow-api.example.com/data', 3000);
15.   console.log(data);
16. } catch (err) {
17.   console.log('使用缓存数据');
18. }

复制代码

特点：只取最先完成的那个结果（成功或失败）。常用于超时控制、多源竞速。

4. Promise.any——多 CDN 资源容灾加载
场景：前端静态资源部署在三个 CDN 上，只要任意一个 CDN 返回成功，就使用该资源，忽略失败的 CDN。

2. async function loadScriptFromCDNs(urls) {
3.   const fetchTasks = urls.map(url => fetch(url).then(res => {
4.     if (!res.ok) throw new Error(`HTTP ${res.status}`);
5.     return res.text();
6.   }));
7.   try {
8.     const scriptContent = await Promise.any(fetchTasks);
9.     eval(scriptContent); // 实际项目中建议使用更安全的方式
10.     console.log('脚本加载成功');
11.   } catch (aggregateError) {
12.     console.error('所有 CDN 均不可用', aggregateError.errors);
13.   }
14. }
15. loadScriptFromCDNs([
16.   'https://cdn1.example.com/lib.js',
17.   'https://cdn2.example.com/lib.js',
18.   'https://cdn3.example.com/lib.js'
19. ]);

复制代码

特点：只要有一个成功就返回，全部失败才抛出异常。非常适合冗余容灾设计。

5. 事件计数器——旧式多文件写入完成后合并压缩
场景：维护一个老项目（基于回调风格），需要等三个日志文件全部写入磁盘后，再执行合并压缩操作。

2. const EventEmitter = require('events');
3. const fs = require('fs');
4. class FileWriter extends EventEmitter {
5.   writeAndNotify(file, data) {
6.     fs.writeFile(file, data, (err) => {
7.       if (err) this.emit('error', err);
8.       else this.emit('done', file);
9.     });
10.   }
11. }
12. const writer = new FileWriter();
13. let completed = 0;
14. const total = 3;
15. function onAllDone() {
16.   console.log('所有文件写入完成，开始合并压缩');
17. }
18. writer.on('done', (file) => {
19.   console.log(`${file} 写入完成`);
20.   if (++completed === total) onAllDone();
21. });
22. writer.writeAndNotify('log1.txt', 'data1');
23. writer.writeAndNotify('log2.txt', 'data2');
24. writer.writeAndNotify('log3.txt', 'data3');

复制代码

特点：原始但可控，适合无法使用 Promise 的旧环境或需要细粒度事件监听时使用。

6. 流式处理——实时聚合多个传感器数据流
场景：物联网网关接收温度、湿度、气压三个传感器的实时数据流，需要每收到一组（三个传感器各一个值）就计算平均值并推送。

2. const { fromEvent, merge, bufferCount } = require('rxjs');
3. const { EventEmitter } = require('events');
4. const sensorA = new EventEmitter();
5. const sensorB = new EventEmitter();
6. const sensorC = new EventEmitter();
7. setInterval(() => sensorA.emit('data', Math.random() * 30), 1000);
8. setInterval(() => sensorB.emit('data', Math.random() * 60), 1000);
9. setInterval(() => sensorC.emit('data', Math.random() * 10), 1000);
10. const streamA = fromEvent(sensorA, 'data');
11. const streamB = fromEvent(sensorB, 'data');
12. const streamC = fromEvent(sensorC, 'data');
13. merge(streamA, streamB, streamC)
14.   .pipe(bufferCount(3))
15.   .subscribe(values => {
16.     const avg = values.reduce((a, b) => a + b, 0) / values.length;
17.     console.log(`实时平均传感器值: ${avg.toFixed(2)}`);
18.   });

复制代码

特点：适合结果逐步产生、需要实时响应的场景。RxJS 提供了强大的组合能力。

7. 队列控制并发——限制同时上传文件的数量
场景：用户一次选择了 100 个文件上传到云存储，必须控制同时上传的并发数为 5，避免网络拥塞和服务器压力过大。

2. const pLimit = require('p-limit');
3. const fs = require('fs').promises;
4. const path = require('path');
5. async function uploadFile(filePath) {
6.   console.log(`开始上传 ${path.basename(filePath)}`);
7.   await new Promise(r => setTimeout(r, 1000)); // 模拟上传
8.   console.log(`完成上传 ${path.basename(filePath)}`);
9.   return filePath;
10. }
11. async function uploadAll(filePaths) {
12.   const limit = pLimit(5); // 最多5个并发
13.   const tasks = filePaths.map(filePath =>
14.     limit(() => uploadFile(filePath))
15.   );
16.   const results = await Promise.all(tasks);
17.   console.log(`全部上传完成，共 ${results.length} 个文件`);
18. }
19. const files = Array.from({ length: 100 }, (_, i) => `/tmp/file${i}.txt`);
20. uploadAll(files);

复制代码

特点：既保证并发效率，又避免资源耗尽。配合 Promise.all 可以等待所有任务完成。

总结：
- 所有任务必须全部成功，结果一起使用 → Promise.all
- 容忍部分失败，但需要知道每个任务的状态 → Promise.allSettled
- 只取最快结果（如超时、多源竞速） → Promise.race
- 只要有一个成功即可，忽略失败 → Promise.any
- 旧项目回调风格或需要细粒度控制 → 事件计数器 / EventEmitter
- 结果流式输出、复杂组合（如传感器数据） → RxJS / 异步迭代器
- 大量任务且需控制并发数量 → 队列 + p-limit

在实际项目中，90% 的异步协作需求都可以用 Promise.all 和 Promise.allSettled 解决。对于更复杂的场景，再考虑流式处理或队列控制。掌握这些模式，你的 Node.js 异步编程能力将更上一层楼。

热心网友5

感谢楼主分享这么实用的总结！正好最近在处理批量上传的任务，`Promise.allSettled` 和 `Promise.any` 的例子对我很有启发，尤其是多 CDN 容灾那个思路，之前一直用 `Promise.race` 但发现失败就会中断，没想到 `any` 更符合容灾场景。另外事件计数器的例子也勾起了我对回调风格的回忆，现在用 Promise 确实清晰多了。楼主有没有遇到过 `Promise.all` 中某个任务抛出非 Error 类型异常的情况？比如 reject 了一个普通字符串，结果处理起来有点坑。