
本文介绍一种轻量、可控的并发优化方案:通过分离单条数据处理逻辑 + 非阻塞式 promise 发起 + 可调延迟节流,实现对数百甚至上千条新闻文章的稳定、不崩溃的并行处理,避免资源过载与服务限流。
本文介绍一种轻量、可控的并发优化方案:通过分离单条数据处理逻辑 + 非阻塞式 promise 发起 + 可调延迟节流,实现对数百甚至上千条新闻文章的稳定、不崩溃的并行处理,避免资源过载与服务限流。
在处理大规模数组(如 100–1000+ 条新闻)时,原始代码中使用 for...of + await 的串行执行模式会带来严重瓶颈:每篇文章需依次完成爬取 → GPT 分析 → DALL·E 制图 → S3 上传 → Strapi 发布共 5 个耗时 I/O 步骤,总耗时呈线性增长(1000 条 × 平均 8 秒 ≈ 2.2 小时),且极易触发目标服务(如 API、Scraper、OpenAI、DALL·E)的速率限制或连接超时,导致失败率陡增。
根本优化思路是:解除单条数据内部步骤的串行依赖(必须保留),但打破不同数据之间的执行阻塞(必须解除)。即:让第 2 篇文章不必等待第 1 篇全部完成才开始——只要控制好发起节奏,就能显著提升吞吐量,同时规避集中并发风险。
✅ 推荐方案:节流式并发(Throttled Concurrency)
该方案无需引入 BullMQ、Redis 或复杂队列系统,仅用原生 JavaScript 即可实现生产级稳定性,兼顾性能、可读性与运维友好性。
第一步:抽离单条处理逻辑为独立函数
const processArticle = async (article) => {
const { title, sourceUrl, id, publishedAt } = article;
console.log(`[Article ${id}] Starting processing...`);
try {
const markup = await scraper(sourceUrl);
const data = await askGpt(markup);
const generatedImageUrl = await generateImg(data?.imageDescription);
const s3ImageUrl = await generateImgUrl(generatedImageUrl, title, id);
await createPost(
data?.title,
data?.abstract,
data?.content,
s3ImageUrl,
publishedAt,
data?.categories
);
console.log(`[Article ${id}] ✅ Successfully published.`);
} catch (err) {
console.error(`[Article ${id}] ❌ Failed:`, err.message);
}
};✅ 优势:职责清晰、便于单元测试、错误隔离(单条失败不影响其余)、日志可追溯。
第二步:主流程启用受控并发
const delay = (ms) => new Promise(resolve => setTimeout(resolve, ms));
const fetchAndProcessNews = async (queryString, from) => {
const query = { queryString, from, size: 1 };
try {
console.log('? Fetching news...');
const { articles } = await searchApi.getNews(query);
console.log(`? Fetched ${articles.length} articles.`);
if (!articles?.length) {
console.log('ℹ️ No articles found.');
return;
}
console.log('⚙️ Initiating concurrent processing...');
const promises = [];
for (const article of articles) {
// 立即启动处理,但不 await —— 实现并发
promises.push(processArticle(article));
// 关键节流:每启动一条,暂停 100ms(可根据实际压测调整)
await delay(100);
}
// 等待所有任务最终完成(成功/失败均计入)
await Promise.allSettled(promises);
console.log('✅ All article processing completed.');
} catch (error) {
console.error('? Critical error in fetchAndProcessNews:', error);
}
};? 为什么是 delay(100)?如何调优?
-
100ms是经验起点:它允许约 10 条/秒 的请求发起速率,远低于多数公共 API 的默认限流阈值(如 OpenAI 通常为 3–5 RPM,DALL·E 为 10 RPM)。 -
实测建议:先设为
500ms(2 条/秒),观察各服务响应时间与错误率;逐步缩短至100ms,若出现429 Too Many Requests或ETIMEDOUT,则回调至200–300ms。 - ⚠️ 注意:
delay控制的是「任务发起间隔」,而非「任务执行时间」;每个processArticle内部仍是串行,确保逻辑正确。
? 关键注意事项
-
错误处理必须内聚:
processArticle中使用try/catch包裹全部步骤,防止单条失败中断整个Promise.allSettled流程。 -
日志需带标识符:如
[Article ${id}],否则并发下console.log顺序混乱,无法定位问题。 -
内存安全:
Promise.allSettled(promises)会持有全部 Promise 实例,对超大数组(>5000)建议分批(如每 100 条一批),避免 V8 堆溢出。 -
非替代方案说明:
- ✅
Promise.allSettled:推荐,返回每个 Promise 的status/value/reason,便于统计成功率。 - ❌
Promise.all:任一失败即 reject,不适合容错场景。 - ⚠️
BullMQ:适合需持久化、重试、优先级、分布式调度的长期任务系统,本场景属短时批量作业,过度设计。
- ✅
✅ 总结
面对大规模异步任务,“节流式并发”是最小可行、最高性价比的优化策略:它不增加架构复杂度,不依赖外部中间件,仅通过函数拆分 + await delay() + Promise.allSettled 三步,即可将处理效率提升 5–10 倍(取决于 I/O 并发度),同时保障服务稳定性与可观测性。上线前务必结合真实接口做压测调优,让延迟参数真正匹配你的基础设施能力。

















