Node.js处理超大JSON标准输出的问题及架构优化咨询

在Node.js中使用shell.js执行Unix命令，结合Bull实现任务队列，可传入如下格式的任务数组：

{
    "tasks": [
        { 
            "name": "task1", 
            "command": "curl -XPOST --header \"Authorization: **********\" --header \"Accept: application/json\" --header \"Content-Type: application/json\" \"https://misp.****//attributes/restSearch/json/5.9.178.143/ip-dst\"" 
        },
        { 
            "name": "task2", 
            "command": "cat <input>" 
        }
    ]
}

第一个任务的输出需保存为文件，作为第二个任务的输入。最初尝试用内存处理输出，但因输出体积过大改为文件存储，然而shell.exec的stdout仍会将完整输出加载到内存中，导致Bull Worker执行任务1时内存溢出失败。需要解决该内存问题，并寻求更优的架构与代码设计思路。现有代码如下：

worker.ts #

import { exec } from "child_process";
import myQueue from "./queue";
import shell from 'shelljs';
import fs from 'fs';
import path from 'path';

myQueue.process(async (job) => {
    console.log(`Processing job with id ${job.id}`);
    const tasks = job.data.tasks;
    let prevOutputFile = '';

    for (const [index, task] of tasks.entries()) {
        let command = task.command;

        // 如果命令中有<input>占位符，替换为上一个任务的输出文件
        if (command.includes('<input>')) {
            if (prevOutputFile) {
                command = command.replace('<input>', prevOutputFile);
            } else {
                console.error('没有可用的前置输出文件作为输入');
                break;
            }
        }

        try {
            // 如果不是最后一个任务，将输出写入文件
            if(index < tasks.length - 1) {
                const outputFile = path.join(__dirname, `${task.name}_output.txt`);
                const output = await execAsync(command, outputFile);

                console.log(`Task ${task.name} complete with output: ${output}`);
                prevOutputFile = outputFile;  // 保存输出文件名给下一个任务
            } else {
                const output = await execAsync(command);

                console.log(`Task ${task.name} complete with output: ${output}`);
            }
            
        } catch (error) {
            console.error(`Task ${task.name} failed with error: ${error.message}`);
            break;
        }
    }
});

function execAsync(cmd: string, outputFile?: string): Promise<string> {
    return new Promise<string>((resolve, reject) => {
        shell.exec(cmd, {silent: true, async: true}, (code, stdout, stderr) => {
            if (code !== 0) {
                reject(new Error(stderr));
            } else {
                if(outputFile) {
                    fs.writeFileSync(outputFile, stdout);
                }
                resolve(stdout);
            }
        })
    })
}

server.ts #

import express, { Request, Response } from 'express';
import myQueue from './queue';

const app = express();
app.use(express.json());
const port = 3000;

app.get('/', (req: Request, res: Response) => {
  res.send('Hello, world!');
});

app.post('/taskChain', async (req: Request, res: Response) => {
    const tasks = req.body.tasks;
    const newJob = await myQueue.add({
      tasks: tasks
    })

    res.json({id: newJob.id})
})

app.listen(port, () => {
  console.log(`Server is running at http://localhost:${port}`);
});

queue.ts #

import Bull from "bull";

const myQueue = new Bull('myQueue')

export default myQueue

一、解决stdout内存占用问题 #

核心问题是shell.exec会将命令的完整输出加载到内存后再写入文件，大输出场景下直接导致内存溢出。可通过以下两种方式解决：

1. 让命令直接输出到文件（最简便） #

修改命令，通过Unix重定向>直接将输出写入目标文件，完全绕过Node.js的内存中转：

// 在任务循环中修改命令处理逻辑
if(index < tasks.length - 1) {
    const outputFile = path.join(__dirname, `${task.name}_output.txt`);
    // 给原命令添加重定向，直接写入文件
    const cmdWithRedirect = `${command} > "${outputFile}"`;
    await execAsync(cmdWithRedirect);
    prevOutputFile = outputFile;
    console.log(`Task ${task.name} completed, output saved to ${outputFile}`);
}

同步修改execAsync，移除内存写入逻辑：

function execAsync(cmd: string): Promise<void> {
    return new Promise((resolve, reject) => {
        shell.exec(cmd, {silent: true, async: true}, (code, _, stderr) => {
            if (code !== 0) {
                reject(new Error(stderr));
            } else {
                resolve();
            }
        })
    })
}

这种方式让系统直接处理文件写入，Node.js仅负责触发命令，不会占用额外内存存储大输出。

2. 用child_process.spawn流式处理输出 #

若需要在Node.js中实时处理输出（比如记录日志），用spawn替代shell.exec，通过流将输出写入文件，内存占用始终维持在缓冲区大小：

import { spawn } from 'child_process';
import { createWriteStream } from 'fs';
import os from 'os';
import shellQuote from 'shell-quote'; // 需安装shell-quote包解析复杂命令

function execToFile(cmd: string, outputFile?: string): Promise<string> {
    return new Promise((resolve, reject) => {
        const parsedCmd = shellQuote.parse(cmd);
        const [command, ...args] = parsedCmd;
        const proc = spawn(command, args, { shell: true });
        
        let stdoutBuffer = '';
        let writeStream;
        
        if (outputFile) {
            writeStream = createWriteStream(outputFile);
            proc.stdout.pipe(writeStream);
        } else {
            proc.stdout.on('data', (chunk) => {
                stdoutBuffer += chunk.toString();
            });
        }
        
        proc.stderr.on('data', (chunk) => {
            console.error(chunk.toString());
        });
        
        proc.on('close', (code) => {
            if (writeStream) writeStream.close();
            if (code !== 0) {
                reject(new Error(`Command failed with code ${code}`));
            } else {
                resolve(outputFile ? outputFile : stdoutBuffer);
            }
        });
        
        proc.on('error', (err) => {
            if (writeStream) writeStream.close();
            reject(err);
        });
    });
}

二、架构优化建议 #

1. 拆分任务为独立Bull作业 #

当前将整个任务链放在单个Worker中执行，单点失败会导致整个链中断。可将每个任务拆为独立Bull作业，通过作业依赖实现链式执行：

• 提交第一个任务时记录其作业ID
• 第一个任务完成后，自动提交第二个任务并传入输出文件路径
• 以此类推，直到最后一个任务完成
  每个任务可独立重试，Worker也能并行处理不同任务链的步骤。

2. 临时文件管理 #

• 使用系统临时目录（os.tmpdir()）存储输出文件，避免污染应用目录
• 为每个任务链生成唯一临时目录，便于批量清理
• 在任务链全部完成或失败后，通过Bull的onComplete和onFailed钩子自动删除临时文件

3. 错误处理与重试 #

• 为每个Bull作业配置重试策略（如失败后自动重试3次）
• 任务失败时记录命令、错误信息、时间等结构化日志
• 支持手动重试失败的任务链步骤

4. 资源隔离 #

• 用Docker隔离每个任务的执行环境，避免命令间的资源冲突或安全风险
• 限制每个任务的CPU、内存配额，防止单个任务耗尽系统资源

三、代码设计思路 #

1. 模块化拆分 #

• 任务执行模块：封装命令执行、流式写入文件的逻辑，作为独立工具函数
• 文件管理模块：负责临时目录创建、文件清理、路径生成
• 任务链调度模块：负责解析任务数组、创建依赖的Bull作业、处理任务间参数传递

2. 任务间通信 #

• 用Bull作业的returnValue传递输出文件路径，替代Worker内存中的变量
• 或用Redis存储任务链状态（当前执行步骤、输出文件路径），Worker可随时读取

3. 可扩展性 #

• 支持自定义任务类型（不仅限于shell命令，还可支持Node.js函数）
• 支持任务链的分支、并行执行
• 添加任务执行前的参数校验钩子和执行后的结果处理钩子

4. 监控与日志 #

• 集成Bull监控面板（如bull-board），实时查看任务状态
• 用结构化日志库（如winston、pino）记录任务执行细节，方便排查问题

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