Node.js Express长任务函数管理：避免超时并后台执行

核心方案思路 #

因为长任务不需要立即返回结果，核心思路是将任务从主进程剥离到独立Worker线程，API仅返回任务标识，后续通过标识查询任务状态/结果，彻底避免API超时。

现有代码的问题点 #

• 重复绑定worker.on('message')事件，后续绑定会覆盖前一个，导致第一个回调永远无法执行
• Worker线程中performTask是异步函数，但未等待其完成就发送消息，会导致任务未跑完就被标记为完成
• tasksById存储Worker实例，可能引发内存泄漏，Worker退出后未清理引用
• 不当使用eval: true参数，应直接传入Worker文件路径而非eval模式

优化后的完整实现 #

1. 主进程API代码（api.ts） #

import { v4 as uuidv4 } from 'uuid';
import { Worker } from 'node:worker_threads';
import path from 'path';
import express from 'express';

const router = express.Router();
// 内存Map存储任务状态，生产环境建议用Redis/MongoDB等持久化存储
const tasksById = new Map<string, {
  status: 'running' | 'completed' | 'error';
  started: number;
  error?: string;
}>();

router.post('/run-long-task', (req, res) => {
  const taskId = uuidv4();
  // 初始化任务状态
  tasksById.set(taskId, {
    status: 'running',
    started: Date.now(),
  });

  // 创建Worker线程，传入Worker文件绝对路径
  const worker = new Worker(path.resolve(__dirname, './worker.ts'), {
    workerData: { taskId },
  });

  // 监听Worker发送的消息（区分完成/进度类型）
  worker.on('message', (data) => {
    if (data.type === 'completed') {
      tasksById.set(taskId, {
        ...tasksById.get(taskId)!,
        status: 'completed',
      });
    } else if (data.type === 'progress') {
      // 可选：处理任务进度
      console.log(`任务 ${taskId} 进度：${data.progress}%`);
    }
  });

  // 监听Worker错误
  worker.on('error', (error) => {
    tasksById.set(taskId, {
      ...tasksById.get(taskId)!,
      status: 'error',
      error: error.message,
    });
  });

  // 监听Worker退出
  worker.on('exit', (code) => {
    const task = tasksById.get(taskId);
    if (task?.status === 'running') {
      tasksById.set(taskId, {
        ...task,
        status: 'error',
        error: `Worker异常退出，退出码：${code}`,
      });
    }
    // 移除所有监听，避免内存泄漏
    worker.removeAllListeners();
  });

  // 立即返回任务ID给客户端
  res.json({ taskId });
});

// 新增任务状态查询接口
router.get('/task-status/:taskId', (req, res) => {
  const { taskId } = req.params;
  const task = tasksById.get(taskId);
  if (!task) {
    return res.status(404).json({ error: '任务不存在' });
  }
  res.json(task);
});

export default router;

2. Worker线程代码（worker.ts） #

import { parentPort, workerData } from 'node:worker_threads';

// 模拟耗时任务（替换为实际业务逻辑）
async function performTask(taskId: string) {
  try {
    // 示例：模拟5分钟的任务，按秒拆分进度
    const totalSteps = 300;
    for (let i = 0; i < totalSteps; i++) {
      // 执行实际任务逻辑
      await new Promise(resolve => setTimeout(resolve, 1000));
      // 可选：向主进程发送进度
      parentPort?.postMessage({
        type: 'progress',
        progress: Math.round((i + 1) / totalSteps * 100),
        taskId,
      });
    }
    // 任务完成，通知主进程
    parentPort?.postMessage({ type: 'completed', taskId });
  } catch (error) {
    // 任务出错，通知主进程
    parentPort?.postMessage({
      type: 'error',
      error: (error as Error).message,
      taskId,
    });
  }
}

// 直接启动任务（使用workerData传入的taskId，无需监听message事件）
if (parentPort && workerData.taskId) {
  performTask(workerData.taskId).finally(() => {
    // 任务结束后主动退出Worker
    process.exit(0);
  });
}

关键优化说明 #

• 移除重复事件绑定：统一处理Worker消息，区分完成/进度类型
• 等待异步任务完成：Worker中确保performTask执行完毕再发送状态消息
• 内存泄漏防护：Worker退出后移除所有事件监听，清理无效引用
• 安全使用Worker参数：直接传入文件路径，避免不安全的eval模式
• 补充状态查询：客户端可通过任务ID轮询获取状态，适配长任务场景
• 持久化建议：生产环境替换内存Map为Redis等持久化存储，避免进程重启丢失任务状态

其他可选方案 #

如果需要更可靠的任务执行（如进程崩溃后可恢复），可考虑：

• 消息队列：使用RabbitMQ、Redis Queue等，将任务放入队列由独立Worker进程消费
• 专业任务框架：如BullMQ，自带任务重试、进度跟踪、持久化等功能，适合复杂长任务场景

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