NodeJS如何基于可用内存动态调整@supercharge/promise-pool并发数？

当然可以基于可用内存动态调整并发任务数，既可以手动实现调度逻辑，也可以借助现成工具包简化开发。以下是具体方案：

一、手动实现动态并发控制 #

核心思路是提前获取系统可用内存与待处理文件大小，动态判断是否启动新任务；任务完成后再检查内存状态，启动后续任务。

关键步骤： #

• 用os.freemem()获取系统可用内存，或通过process.memoryUsage()计算进程剩余可用内存
• 提前调用fs.promises.stat()获取每个文件大小，避免读取时才触发内存不足问题
• 维护任务队列，每次启动任务前检查内存是否足够容纳当前文件，不足则等待后重试

代码示例： #

const os = require('os');
const fs = require('fs').promises;

// 单个文件处理逻辑
async function processFile(filePath) {
  const filedata = await fs.readFile(filePath);
  // 替换为你的业务处理逻辑
  console.log(`处理完成：${filePath}`);
}

async function dynamicConcurrentTaskRunner(fileList) {
  // 预获取所有文件的路径与大小
  const fileMetadataList = await Promise.all(
    fileList.map(async path => ({
      path,
      size: (await fs.stat(path)).size
    }))
  );

  let activeTaskCount = 0;
  let currentQueueIndex = 0;

  async function tryStartNextTask() {
    if (currentQueueIndex >= fileMetadataList.length) return;

    const { path, size } = fileMetadataList[currentQueueIndex];
    // 预留200MB内存给系统及其他进程，避免内存耗尽
    const safeAvailableMemory = os.freemem() - 200 * 1024 * 1024;

    if (safeAvailableMemory > size) {
      currentQueueIndex++;
      activeTaskCount++;
      try {
        await processFile(path);
      } catch (err) {
        console.error(`处理文件失败：${path}`, err);
      } finally {
        activeTaskCount--;
        // 当前任务完成后，立即尝试启动下一个
        await tryStartNextTask();
      }
    } else {
      // 内存不足，100ms后再检查
      setTimeout(tryStartNextTask, 100);
    }
  }

  // 先批量启动一批任务，直到内存不足或队列耗尽
  while (currentQueueIndex < fileMetadataList.length) {
    const { size } = fileMetadataList[currentQueueIndex];
    const safeAvailableMemory = os.freemem() - 200 * 1024 * 1024;
    if (safeAvailableMemory > size) {
      tryStartNextTask();
    } else {
      break;
    }
  }
}

// 使用示例
const targetFiles = ['/path/to/file1', '/path/to/file2', ...];
dynamicConcurrentTaskRunner(targetFiles);

二、使用现成工具包 #

有不少并发控制工具支持自定义调度逻辑，可结合内存检查动态调整并发数，推荐以下两个：

1. p-queue #

这是现代Promise队列工具，支持动态修改并发数。可在任务执行前或定期检查内存，调整队列的concurrency参数：

const PQueue = require('p-queue');
const os = require('os');
const fs = require('fs').promises;

// 初始化队列，初始并发数设为1
const taskQueue = new PQueue({ concurrency: 1 });

// 根据文件大小与可用内存调整并发数
async function adjustConcurrency(fileSize) {
  const safeAvailableMemory = os.freemem() - 200 * 1024 * 1024;
  // 计算当前可同时处理的最大文件数
  const maxConcurrency = Math.floor(safeAvailableMemory / fileSize);
  // 确保并发数至少为1
  taskQueue.concurrency = Math.max(1, maxConcurrency);
}

async function processFile(filePath) {
  const { size } = await fs.stat(filePath);
  // 处理前调整并发数
  await adjustConcurrency(size);

  const filedata = await fs.readFile(filePath);
  // 业务处理逻辑
}

// 将所有文件任务加入队列
const targetFiles = ['/path/to/file1', '/path/to/file2', ...];
targetFiles.forEach(file => taskQueue.add(() => processFile(file)));

2. async.queue #

async库中的queue方法也支持自定义任务调度，通过控制concurrency参数结合内存检查，实现动态并发控制。

三、更根本的优化建议 #

如果场景允许，用文件流替代一次性读取整个文件是解决内存问题的最优方案。通过fs.createReadStream分块读取处理，内存占用会大幅降低，无需担心并发数导致的内存峰值：

const fs = require('fs');

function processFileWithStream(filePath) {
  return new Promise((resolve, reject) => {
    const stream = fs.createReadStream(filePath);
    stream.on('data', chunk => {
      // 分块处理文件内容
    });
    stream.on('end', resolve);
    stream.on('error', reject);
  });
}

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