如何在Node.js中借助Busboy正确停止流并删除文件？

Multer 基于 Busboy 构建，其核心逻辑集中在make-middleware.js文件中，核心代码如下：

var is = require('type-is')
var Busboy = require('busboy')
var extend = require('xtend')
var onFinished = require('on-finished')
var appendField = require('append-field')

var Counter = require('./counter')
var MulterError = require('./multer-error')
var FileAppender = require('./file-appender')
var removeUploadedFiles = require('./remove-uploaded-files')

function drainStream (stream) {
  stream.on('readable', stream.read.bind(stream))
}

function makeMiddleware (setup) {
  return function multerMiddleware (req, res, next) {
    if (!is(req, ['multipart'])) return next()

    var options = setup()

    var limits = options.limits
    var storage = options.storage
    var fileFilter = options.fileFilter
    var fileStrategy = options.fileStrategy
    var preservePath = options.preservePath

    req.body = Object.create(null)

    var busboy

    try {
      busboy = new Busboy({ headers: req.headers, limits: limits, preservePath: preservePath })
    } catch (err) {
      return next(err)
    }

    var appender = new FileAppender(fileStrategy, req)
    var isDone = false
    var readFinished = false
    var errorOccured = false
    var pendingWrites = new Counter()
    var uploadedFiles = []

    function done (err) {
      if (isDone) return
      isDone = true

      req.unpipe(busboy)
      drainStream(req)
      busboy.removeAllListeners()

      onFinished(req, function () { next(err) })
    }

    function indicateDone () {
      if (readFinished && pendingWrites.isZero() && !errorOccured) done()
    }

    function abortWithError (uploadError) {
      if (errorOccured) return
      errorOccured = true

      pendingWrites.onceZero(function () {
        function remove (file, cb) {
          storage._removeFile(req, file, cb)
        }

        removeUploadedFiles(uploadedFiles, remove, function (err, storageErrors) {
          if (err) return done(err)

          uploadError.storageErrors = storageErrors
          done(uploadError)
        })
      })
    }

    function abortWithCode (code, optionalField) {
      abortWithError(new MulterError(code, optionalField))
    }

    // handle text field data
    busboy.on('field', function (fieldname, value, fieldnameTruncated, valueTruncated) {
      if (fieldname == null) return abortWithCode('MISSING_FIELD_NAME')
      if (fieldnameTruncated) return abortWithCode('LIMIT_FIELD_KEY')
      if (valueTruncated) return abortWithCode('LIMIT_FIELD_VALUE', fieldname)

      // Work around bug in Busboy
      if (limits && Object.prototype.hasOwnProperty.call(limits, 'fieldNameSize')) {
        if (fieldname.length > limits.fieldNameSize) return abortWithCode('LIMIT_FIELD_KEY')
      }

      appendField(req.body, fieldname, value)
    })

    // handle files
    busboy.on('file', function (fieldname, fileStream, filename, encoding, mimetype) {
      // don't attach to the files object, if there is no file
      if (!filename) return fileStream.resume()

      // Work around bug in Busboy
      if (limits && Object.prototype.hasOwnProperty.call(limits, 'fieldNameSize')) {
        if (fieldname.length > limits.fieldNameSize) return abortWithCode('LIMIT_FIELD_KEY')
      }

      var file = {
        fieldname: fieldname,
        originalname: filename,
        encoding: encoding,
        mimetype: mimetype
      }

      var placeholder = appender.insertPlaceholder(file)

      fileFilter(req, file, function (err, includeFile) {
        if (err) {
          appender.removePlaceholder(placeholder)
          return abortWithError(err)
        }

        if (!includeFile) {
          appender.removePlaceholder(placeholder)
          return fileStream.resume()
        }

        var aborting = false
        pendingWrites.increment()

        Object.defineProperty(file, 'stream', {
          configurable: true,
          enumerable: false,
          value: fileStream
        })

        fileStream.on('error', function (err) {
          pendingWrites.decrement()
          abortWithError(err)
        })

        fileStream.on('limit', function () {
          aborting = true
          abortWithCode('LIMIT_FILE_SIZE', fieldname)
        })

        storage._handleFile(req, file, function (err, info) {
          if (aborting) {
            appender.removePlaceholder(placeholder)
            uploadedFiles.push(extend(file, info))
            return pendingWrites.decrement()
          }

          if (err) {
            appender.removePlaceholder(placeholder)
            pendingWrites.decrement()
            return abortWithError(err)
          }

          var fileInfo = extend(file, info)

          appender.replacePlaceholder(placeholder, fileInfo)
          uploadedFiles.push(fileInfo)
          pendingWrites.decrement()
          indicateDone()
        })
      })
    })

    busboy.on('error', function (err) { abortWithError(err) })
    busboy.on('partsLimit', function () { abortWithCode('LIMIT_PART_COUNT') })
    busboy.on('filesLimit', function () { abortWithCode('LIMIT_FILE_COUNT') })
    busboy.on('fieldsLimit', function () { abortWithCode('LIMIT_FIELD_COUNT') })
    busboy.on('finish', function () {
      readFinished = true
      indicateDone()
    })

    req.pipe(busboy)
  }
}

module.exports = makeMiddleware

当前实现的核心逻辑可概括为：

abortWithError(() => pendingWrites.onceZero(() => removeFiles()...))

这种方式会等待所有文件写入磁盘完成后再执行删除操作，在大文件上传场景下会造成明显的资源浪费。如果能直接获取并终止磁盘写入流，就能大幅节省资源，但需要掌握正确的流终止方式。

数据流的起始逻辑是：

req.pipe(busboy)

请求流被导入Busboy解析实例，该实例会触发file和field事件。file事件携带的fileStream通常会被管道到磁盘写入流，示例用法如下：

fileStream.on('error', function (err) {
  pendingWrites.decrement()
  abortWithError(err)
})

fileStream.on('limit', function () {
  aborting = true
  abortWithCode('LIMIT_FILE_SIZE', fieldname)
})

const writeStream = fs.createWriteStream('/tmp/blah')
fileStream.pipe(writeStream)

Multer内部无法获取到这个writeStream，这就是现有实现要等待写入完成再删除的原因。但如果能拿到写入流，该如何操作？以下是方案分析与最佳实践：

方案分析 #

• writeStream.close()：该方法会等待缓冲区所有数据写入磁盘后再关闭流，无法实现立即终止的效果，不适合快速终止场景。
• writeStream.destroy()：该方法会立即终止流，忽略缓冲区剩余数据，同时触发error事件（可监听并忽略预期错误），是快速终止写入的有效方式。
• fileStream.unpipe(writeStream)：仅解除管道不会停止写入流本身，写入流仍会处理已接收的数据，单独使用不足以终止写入。
• 立即执行fs.unlink('/tmp/blah')：流未终止时删除文件会导致写入失败并抛出错误，属于非优雅操作，必须等流完全终止后再执行删除。

正确的组合方案 #

正确的操作流程是先销毁写入流，解除管道，待流完全关闭后删除文件，完整示例如下：

const fs = require('fs')
const writeStream = fs.createWriteStream('/tmp/blah')
let fileDeleted = false

// 处理文件流错误
fileStream.on('error', function (err) {
  // 立即销毁写入流，终止写入
  writeStream.destroy()
  // 解除管道，防止后续数据流入
  fileStream.unpipe(writeStream)
  // 等待写入流关闭后删除文件
  writeStream.on('close', () => {
    if (!fileDeleted) {
      fileDeleted = true
      fs.unlink('/tmp/blah', (unlinkErr) => {
        if (unlinkErr) console.error('删除临时文件失败:', unlinkErr)
        pendingWrites.decrement()
        abortWithError(err)
      })
    }
  })
})

// 处理文件大小超限
fileStream.on('limit', function () {
  aborting = true
  // 执行相同的终止流程
  writeStream.destroy()
  fileStream.unpipe(writeStream)
  writeStream.on('close', () => {
    if (!fileDeleted) {
      fileDeleted = true
      fs.unlink('/tmp/blah', (unlinkErr) => {
        if (unlinkErr) console.error('删除临时文件失败:', unlinkErr)
        abortWithCode('LIMIT_FILE_SIZE', fieldname)
      })
    }
  })
})

fileStream.pipe(writeStream)

额外注意事项 #

1. 必须监听写入流的close事件再执行删除操作，确保流已完全终止，避免文件被占用时删除引发错误。
2. 标记fileDeleted变量防止重复删除，避免多次触发事件导致重复调用fs.unlink。
3. 销毁流时会触发error事件，需提前监听并处理（如忽略预期内的终止错误），避免未捕获异常。

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