Node.js中多异步进程向同一文件追加写入的正确方法

原代码的问题 #

你当前的实现存在两个核心问题：

1. 阻塞事件循环：fs.appendFileSync是同步IO操作，会暂停整个Node.js事件循环，导致其他异步任务无法执行，完全失去了异步并发的优势。
2. 潜在写入混乱：单线程下同步写入会串行执行，但如果后续调整为异步写入，多个并发操作直接操作同一文件描述符，会出现内容交错（比如A任务写了一半，B任务写入就开始执行）。

优化方案 #

场景1：同一进程内的多个异步任务 #

这是你示例代码对应的场景，推荐两种实现方式：

方式1：异步写入 + 串行化队列

通过维护一个Promise队列，让所有写入操作串行执行，既保证异步IO不阻塞事件循环，又避免写入内容错乱。

const fs = require('fs').promises;
// 初始化写入队列
let writeQueue = Promise.resolve();

// 封装串行写入函数
function queueWrite(fd, data) {
  writeQueue = writeQueue
    .then(async () => {
      await fs.write(fd, `${data}\n`, 'utf8');
    })
    .catch(err => {
      console.error('写入失败:', err);
    });
  return writeQueue;
}

// 修改异步任务函数
async function myAsyncProcess(thing, fileDescriptor) {
  const changedThing = await doSomething(thing);
  // 加入写入队列
  await queueWrite(fileDescriptor, changedThing);
  // 其他业务逻辑
}

// 主流程
async function main() {
  const pathToFile = './output.txt';
  const severalThings = ['item1', 'item2', 'item3'];

  // 打开文件获取描述符
  const fileDescriptor = await fs.open(pathToFile, 'a');
  // 启动所有异步任务
  const allTheAsyncJobs = severalThings.map(thing => myAsyncProcess(thing, fileDescriptor));
  const results = await Promise.allSettled(allTheAsyncJobs);
  
  // 记得关闭文件描述符
  await fileDescriptor.close();
  
  // 处理任务结果
  results.forEach((result, index) => {
    if (result.status === 'rejected') {
      console.error(`任务${index}失败:`, result.reason);
    }
  });
}

// 模拟异步处理逻辑
async function doSomething(thing) {
  await new Promise(resolve => setTimeout(resolve, 100));
  return `processed_${thing}`;
}

main().catch(err => console.error('主流程失败:', err));

方式2：使用文件流（更推荐）

Node.js的文件流自带背压机制，能自动平衡写入速度和磁盘IO速度，避免内存溢出，同时内部处理了写入的并发问题，代码更简洁高效。

const fs = require('fs');

// 主流程
async function main() {
  const pathToFile = './output.txt';
  const severalThings = ['item1', 'item2', 'item3'];

  // 创建追加模式的写入流
  const writeStream = fs.createWriteStream(pathToFile, { flags: 'a' });
  // 监听流错误
  writeStream.on('error', err => console.error('流写入错误:', err));

  // 封装异步流写入函数，处理背压
  function streamWrite(data) {
    return new Promise((resolve, reject) => {
      // 写入数据
      const canContinue = writeStream.write(`${data}\n`, 'utf8', err => {
        if (err) reject(err);
      });
      // 如果流缓冲区满了，等待drain事件再继续
      if (!canContinue) {
        writeStream.once('drain', resolve);
      } else {
        resolve();
      }
    });
  }

  // 异步任务函数
  async function myAsyncProcess(thing) {
    const changedThing = await doSomething(thing);
    await streamWrite(changedThing);
    // 其他业务逻辑
  }

  // 启动所有任务
  const allTheAsyncJobs = severalThings.map(thing => myAsyncProcess(thing));
  const results = await Promise.allSettled(allTheAsyncJobs);
  
  // 结束流，等待所有写入完成
  await new Promise(resolve => writeStream.end(resolve));
  
  // 处理结果
  results.forEach((result, index) => {
    if (result.status === 'rejected') {
      console.error(`任务${index}失败:`, result.reason);
    }
  });
}

// 模拟异步处理
async function doSomething(thing) {
  await new Promise(resolve => setTimeout(resolve, 100));
  return `processed_${thing}`;
}

main().catch(err => console.error('主流程失败:', err));

场景2：多个独立的Node.js进程 #

如果是真正的多进程（比如用cluster或child_process创建的进程），共用文件描述符需要跨进程传递，同时必须用文件锁保证写入原子性：

父进程代码

const fs = require('fs');
const { fork } = require('child_process');
const pathToFile = './output.txt';

async function main() {
  const severalThings = ['item1', 'item2', 'item3'];
  // 父进程打开文件
  const fd = fs.openSync(pathToFile, 'a');

  // 启动子进程并传递文件描述符
  const childPromises = severalThings.map(thing => {
    return new Promise((resolve, reject) => {
      const child = fork('./child-process.js');
      // 传递数据和文件描述符
      child.send({ thing, fd }, { fd });
      // 监听子进程退出
      child.on('exit', code => {
        if (code === 0) resolve();
        else reject(new Error(`子进程退出码: ${code}`));
      });
      child.on('error', reject);
    });
  });

  await Promise.allSettled(childPromises);
  // 父进程关闭文件描述符
  fs.closeSync(fd);
}

main().catch(err => console.error('主流程失败:', err));

子进程代码（child-process.js）

const fs = require('fs').promises;
const lockfile = require('proper-lockfile'); // 需要先安装：npm install proper-lockfile

process.on('message', async ({ thing, fd }) => {
  try {
    // 获取文件锁，避免多进程写入冲突
    const releaseLock = await lockfile.lock(fd, { lockfilePath: './output.lock' });
    
    // 异步处理数据
    const changedThing = await doSomething(thing);
    // 写入文件
    await fs.write(fd, `${changedThing}\n`, 'utf8');
    
    // 释放锁
    await releaseLock();
    process.exit(0);
  } catch (err) {
    console.error('子进程写入失败:', err);
    process.exit(1);
  }
});

async function doSomething(thing) {
  await new Promise(resolve => setTimeout(resolve, 100));
  return `processed_${thing}`;
}

关键注意事项 #

• 同一进程内：避免使用同步IO，优先用异步写入+队列或文件流，保证并发处理的同时不阻塞事件循环。
• 多进程场景：必须使用文件锁，否则会出现写入内容交错的问题；文件描述符需要由父进程打开后传递给子进程，不要让每个子进程单独打开文件。

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