使用BullMQ沙箱Worker从GCS下载大文件触发任务停滞问题

问题分析 #

沙箱Worker的停滞错误并非仅由CPU密集型操作触发，即使是IO密集的大文件下载，也可能因为以下原因导致：

• 同步IO操作阻塞了沙箱进程的事件循环，无法及时响应主进程的心跳检测
• 大文件一次性加载导致内存过载，引发频繁GC或进程卡顿
• 系统CPU/内存资源被占满，沙箱进程无法完成必要的簿记操作

解决方案建议 #

1. 全程使用异步IO操作，避免阻塞事件循环 #

确保GCS下载和本地文件写入都使用异步API，杜绝同步方法（如fs.writeFileSync）。以@google-cloud/storage为例：

const { Storage } = require('@google-cloud/storage');
const fs = require('fs').promises;

async function processor(job) {
  const storage = new Storage();
  const bucket = storage.bucket(job.data.bucketName);
  const file = bucket.file(job.data.filePath);
  
  // 异步下载文件
  await file.download({ destination: job.data.localPath });
  
  // 后续分析操作也保持异步
  await analyzeFile(job.data.localPath);
}

2. 流式下载大文件，降低内存占用 #

对于10GB以上的文件，流式传输可避免一次性将整个文件加载到内存，大幅减少内存压力：

const { Storage } = require('@google-cloud/storage');
const fs = require('fs');

async function processor(job) {
  const storage = new Storage();
  const bucket = storage.bucket(job.data.bucketName);
  const file = bucket.file(job.data.filePath);
  
  return new Promise((resolve, reject) => {
    const writeStream = fs.createWriteStream(job.data.localPath);
    file.createReadStream()
      .pipe(writeStream)
      .on('finish', resolve)
      .on('error', reject);
  });
}

3. 调整BullMQ的停滞检测参数 #

根据大文件下载的耗时，适当增大停滞检测的间隔和最大重试次数，给任务足够的执行时间：

const { Worker } = require('bullmq');

const worker = new Worker('your-queue', './processor.js', {
  sandbox: true,
  stalledInterval: 300000, // 5分钟，默认30秒
  maxStalledCount: 3, // 允许3次停滞检测失败，默认1次
});

4. 定期发送任务心跳，主动告知主进程状态 #

在下载过程中，通过job.progress()或job.update()定期更新任务状态，让主进程明确任务仍在运行：

async function processor(job) {
  const storage = new Storage();
  const bucket = storage.bucket(job.data.bucketName);
  const file = bucket.file(job.data.filePath);
  const fileSize = (await file.getMetadata())[0].size;
  
  let downloadedBytes = 0;
  
  return new Promise((resolve, reject) => {
    const writeStream = fs.createWriteStream(job.data.localPath);
    file.createReadStream()
      .on('data', async (chunk) => {
        downloadedBytes += chunk.length;
        // 每下载10%更新一次进度
        if (Math.floor((downloadedBytes / fileSize) * 10) % 1 === 0) {
          await job.progress(Math.round((downloadedBytes / fileSize) * 100));
        }
      })
      .pipe(writeStream)
      .on('finish', resolve)
      .on('error', reject);
  });
}

5. 限制并发任务数，避免系统资源过载 #

如果服务器资源有限，减少Worker的并发数，避免同时下载多个大文件导致资源耗尽：

const worker = new Worker('your-queue', './processor.js', {
  sandbox: true,
  concurrency: 1, // 一次只处理一个任务
});

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