Node.js Fork子进程未退出及批量异步下载问题求助

问题根源分析 #

1. 子进程未退出：你的download函数标记为async但内部未返回Promise，await完全无效，子进程不会等待下载完成就继续运行；且单个子进程处理多个任务时，调用process.exit()会直接终止整个子进程，导致所有未完成任务中断。
2. 批次等待失效：当前仅创建一个子进程并批量发送任务，无法追踪任务完成状态，自然无法实现“完成一批再启动下一批”的逻辑。

修正方案与代码实现 #

子进程代码（a.js）

改为单任务单进程模式，修复异步逻辑并增加错误处理：

const down_path = 'test/';
const https = require('https');
const fs = require('fs');
const path = require('path');

// 确保下载目录存在
if (!fs.existsSync(down_path)) {
  fs.mkdirSync(down_path, { recursive: true });
}

process.on('message', async (arr) => {
  console.log("CHILD: 收到下载任务", arr);
  const [url, name, type, year] = arr;
  try {
    await download(url, name, type, year);
    console.log(`下载完成: ${year}`);
    process.send({ status: 'success', year });
    // 单任务完成后退出当前子进程
    process.exit(0);
  } catch (err) {
    console.error(`下载失败 ${year}:`, err);
    process.send({ status: 'error', year, error: err.message });
    process.exit(1);
  }
});

// 封装下载逻辑为Promise，确保await生效
function download(url, name, type, year) {
  return new Promise((resolve, reject) => {
    const filePath = path.join(down_path, `${name}_${type}_${year}.pdf`);
    const stream = fs.createWriteStream(filePath);

    https.get(url, (res) => {
      // 处理HTTP错误状态码
      if (res.statusCode !== 200) {
        reject(new Error(`请求失败，状态码: ${res.statusCode}`));
        stream.close();
        return;
      }

      res.pipe(stream);

      stream.on('finish', () => {
        stream.close();
        resolve();
      });

      stream.on('error', (err) => {
        reject(err);
      });
    }).on('error', (err) => {
      reject(err);
    });
  });
}

主进程代码

实现批次控制，等待当前批次所有子进程完成后再启动下一批：

const fork = require('child_process').fork;
const path = require('path');

// 测试URL列表（可自行扩展）
const url_s = [
  "https://www.responsibilityreports.com/HostedData/ResponsibilityReportArchive/a/NASDAQ_AAON_2020.pdf",
  "https://www.responsibilityreports.com/HostedData/ResponsibilityReportArchive/a/NASDAQ_AAON_2019.pdf",
  "https://www.responsibilityreports.com/HostedData/ResponsibilityReportArchive/a/NASDAQ_AAPL_2022.pdf",
  "https://www.responsibilityreports.com/HostedData/ResponsibilityReportArchive/a/NASDAQ_AAPL_2021.pdf",
  "https://www.responsibilityreports.com/HostedData/ResponsibilityReportArchive/a/NASDAQ_AAPL_2020.pdf",
  "https://www.responsibilityreports.com/HostedData/ResponsibilityReportArchive/a/NYSE_A_2022.pdf",
  "https://www.responsibilityreports.com/HostedData/ResponsibilityReportArchive/a/NYSE_A_2021.pdf"
];
const year_s = ["2020","2019","2022","2021","2020","2022","2021"];
const name = 'a';
const type = 'a';

// 每次批量下载数量
const BATCH_SIZE = 5;
let currentBatchIndex = 0;

// 处理单个批次
async function processBatch() {
  if (currentBatchIndex >= url_s.length) {
    console.log("所有下载任务已完成");
    return;
  }

  // 获取当前批次任务
  const batchStart = currentBatchIndex;
  const batchEnd = Math.min(currentBatchIndex + BATCH_SIZE, url_s.length);
  const batchTasks = url_s.slice(batchStart, batchEnd).map((url, idx) => ({
    url,
    year: year_s[batchStart + idx],
    name,
    type
  }));

  console.log(`启动第${Math.floor(batchStart/BATCH_SIZE)+1}批次，共${batchTasks.length}个任务`);

  // 启动子进程并收集完成Promise
  const batchPromises = batchTasks.map(task => {
    return new Promise((resolve) => {
      const child = fork(path.join(__dirname, 'a.js'));
      
      child.send([task.url, task.name, task.type, task.year]);

      child.on('message', (msg) => {
        msg.status === 'success' 
          ? console.log(`任务完成: ${msg.year}`) 
          : console.error(`任务失败: ${msg.year} - ${msg.error}`);
      });

      child.on('exit', (code) => {
        console.log(`子进程退出，退出码: ${code}`);
        resolve();
      });
    });
  });

  // 等待当前批次所有任务完成
  await Promise.all(batchPromises);
  console.log(`第${Math.floor(batchStart/BATCH_SIZE)+1}批次处理完成`);

  // 处理下一批次
  currentBatchIndex = batchEnd;
  processBatch();
}

// 启动第一个批次
processBatch();

关键改进点 #

1. 子进程逻辑：

   • 将下载逻辑封装为Promise，让await真正生效，确保子进程等待下载完成后再退出。
   • 单任务单进程模式，process.exit()仅终止当前任务的子进程，不会影响其他任务。
   • 增加HTTP状态码、流错误等异常处理，避免子进程异常挂起。

2. 主进程逻辑：

   • 实现批次控制，每次启动指定数量的子进程。
   • 使用Promise.all等待当前批次所有子进程退出，严格保证“完成一批再启动下一批”的逻辑。
   • 增加进度日志，方便追踪任务状态。

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