Node.js中确保目录存在的createWriteStream实现优化问题

我原本的实现如下：

const createWritable = async (path) => {
  // 当目录不存在时递归创建
  await fsp.mkdir(dirname(path), { recursive: true });
  // 创建Node.js可写流供使用
  return fs.createWriteStream(path);
};

但这个实现必须通过const writable = await createWritable(path)获取流，我尝试把异步目录创建整合到流内部，写出了下面的代码，测试时偶尔会出现ENOENT: no such file or directory, open '...'错误：

const createWritable = async (path) => {
  let writer;
  return new Writable({
    async construct(next) {
      await fsp.mkdir(dirname(path), { recursive: true });
      writer = createWriteStream(path);
      next();
    },
    write(chunk, encoding, next) {
      writer.encoding = encoding;
      writer.write(chunk);
      next();
    },
  });
}

疑问解答 #

1. 是否需要设置writer.encoding = encoding？ #

不需要。fs.createWriteStream的write方法本身就支持传入encoding参数，直接调用writer.write(chunk, encoding)即可。手动设置writer.encoding属于冗余操作，还可能在多chunk写入场景下覆盖默认编码，引发不必要的问题。

2. 是否缺少错误处理？会不会是ENOENT的原因？ #

错误处理确实缺失，但偶尔出现的ENOENT更可能是时序问题：虽然Node.js的Writable默认会在construct钩子完成后才处理写入，但如果测试中存在同步触发写入的逻辑，可能导致目录创建完成前就执行了createWriteStream。

另外，你的write方法没有处理writer.write的返回值，也没有监听writer的错误事件，一旦writer出错，错误会被静默吞掉，导致流状态异常。

3. 有没有更优的实现方式？ #

有两种更简洁可靠的方案：

方案一：用pipeline结合异步前置操作

无需手动封装Writable，直接用stream.pipeline先完成目录创建，再将数据导向标准的createWriteStream，pipeline会自动处理流的错误和关闭逻辑：

const { pipeline } = require('stream/promises');
const { createWriteStream } = require('fs');
const { mkdir } = require('fs/promises');
const { dirname } = require('path');

async function writeToPath(readable, path) {
  await mkdir(dirname(path), { recursive: true });
  await pipeline(readable, createWriteStream(path));
}

使用时直接传入可读流即可，稳定性远高于手动封装的流。

方案二：封装带就绪等待的可写流

如果需要返回可写流对象，可以通过就绪Promise确保所有写入操作等待目录创建完成后再执行，同时完善错误处理：

const { finished } = require('stream');
const { createWriteStream } = require('fs');
const { mkdir } = require('fs/promises');
const { dirname } = require('path');

function createWritable(path) {
  let innerStream;
  // 先执行目录创建，确保流创建前目录存在
  const readyPromise = mkdir(dirname(path), { recursive: true })
    .then(() => {
      innerStream = createWriteStream(path);
    })
    .catch(err => {
      if (innerStream) innerStream.destroy(err);
      throw err;
    });

  return new Writable({
    async write(chunk, encoding, callback) {
      await readyPromise;
      innerStream.write(chunk, encoding, callback);
    },
    final(callback) {
      readyPromise.then(() => innerStream.end(callback)).catch(callback);
    },
    destroy(err, callback) {
      if (innerStream) innerStream.destroy(err);
      readyPromise.catch(() => {}).finally(() => callback(err));
    }
  });
}

这个实现彻底避免了时序问题，同时覆盖了流的错误、关闭、销毁等全生命周期逻辑。

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