Node.js如何将接收的音频流分片写入多个单文件100MB的文件

原方案不生效的原因 #

当你使用.pipe()将可读流绑定到fs.createWriteStream实例后，可写流的写入权限完全由管道机制接管，所有写入操作都是自动调度的，此时你手动调用fs.truncate()、fs.write()会和内置的写入逻辑产生时序冲突，自然无法生效。

实现思路 #

放弃直接使用pipe()绑定单个可写流，改为手动监听可读流的data事件，自行统计每个分片的已写入字节数，达到100MB阈值时自动切换新的写入文件即可。

完整实现代码 #

const fs = require('fs')
const path = require('path')

// 配置项
const CHUNK_SIZE = 100 * 1024 * 1024 // 每个分片100MB
const OUTPUT_DIR = './audio_chunks' // 分片存储目录
const BASE_FILENAME = 'audio_part' // 分片文件名前缀

// 初始化目录
if (!fs.existsSync(OUTPUT_DIR)) {
  fs.mkdirSync(OUTPUT_DIR, { recursive: true })
}

// 状态变量
let currentFileIndex = 1
let currentWriteStream = null
let bytesWrittenToCurrent = 0

// 创建新的分片写入流
function createNewWriteStream() {
  if (currentWriteStream) {
    currentWriteStream.end()
  }
  const filename = path.join(OUTPUT_DIR, `${BASE_FILENAME}_${currentFileIndex}.raw`)
  currentWriteStream = fs.createWriteStream(filename)
  bytesWrittenToCurrent = 0
  currentFileIndex++
  // 处理背压
  currentWriteStream.on('drain', () => {
    stream.resume() // 之前如果暂停了流，这里恢复
  })
  currentWriteStream.on('error', (err) => {
    console.error('写入流错误:', err)
    stream.destroy(err)
  })
}

// 初始化第一个写入流
createNewWriteStream()

// 你的原始音频输入流，这里替换为你实际的流对象
// const stream = 你接收到的音频buffer流
stream.on('data', (chunk) => {
  let remainingChunk = chunk
  while (remainingChunk.length > 0) {
    const availableSpace = CHUNK_SIZE - bytesWrittenToCurrent
    if (remainingChunk.length <= availableSpace) {
      // 当前分片还有足够空间，直接写入
      const canWrite = currentWriteStream.write(remainingChunk)
      bytesWrittenToCurrent += remainingChunk.length
      remainingChunk = Buffer.alloc(0)
      if (!canWrite) {
        stream.pause() // 处理背压，暂停读取
      }
    } else {
      // 剩余空间不足，拆分chunk
      const chunkToWrite = remainingChunk.subarray(0, availableSpace)
      const canWrite = currentWriteStream.write(chunkToWrite)
      remainingChunk = remainingChunk.subarray(availableSpace)
      bytesWrittenToCurrent += availableSpace
      // 切换新分片
      createNewWriteStream()
      if (!canWrite) {
        stream.pause()
        // 等drain之后再继续处理剩下的chunk
        currentWriteStream.once('drain', () => stream.resume())
        break
      }
    }
  }
})

// 流读取完成收尾
stream.on('end', () => {
  if (currentWriteStream) {
    currentWriteStream.end()
  }
  console.log(`所有分片写入完成，共生成${currentFileIndex - 1}个文件`)
})

stream.on('error', (err) => {
  console.error('输入流错误:', err)
  if (currentWriteStream) {
    currentWriteStream.destroy()
  }
})

注意事项 #

• 如果你需要存储的是可直接播放的音频格式而非原始buffer，需要额外补充音频头写入逻辑，上述代码存储的是原始二进制分片
• 代码中已经处理了流背压问题，不会因为写入速度跟不上读取速度导致内存溢出
• 可根据实际需求修改分片大小、存储路径、文件名生成规则

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