有推荐的 Nodejs Stream 三方库吗？

V2EX = way to explore

V2EX 是一个关于分享和探索的地方

现在注册

已注册用户请登录

这是一个创建于 510 天前的主题，其中的信息可能已经有所发展或是发生改变。

今天在把一个 golang 写的 bencode 编解码库迁移到 js ，才发现 nodejs 的 stream 是有多难用。。。

ReadStream 的 read 方法太难用，调用 read ，如果读取长度超过了已经准备好的字节数据长度，会直接返回 null(即便剩余文件字节长度远远超过了读取的长度)。

测试用的文件是一个 bt 种子，只在解析 pieces 的时候会出问题，因为 pieces 的长度是 6700 ，但是缓冲区剩余可读字节没有这么多，read(6700)会自己返回 null 。

开始我尝试自己创建 Buffer ，每次只读 1 个字节，用循环处理拼接 Buffer ，后来觉得这样做太操蛋了。

之后看爆栈上的解答，说要监听 readable 事件，因为可能会多次回调。

https://stackoverflow.com/questions/41478192/nodejs-readable-read-return-null

我对 nodejs 的 stream 不是很熟练，我不想写一些 hack 的代码，求问有一些好用的三方 Stream 库吗？

第 1 条附言 · 2023-07-26 16:50:09 +08:00

最终方案：

  async readBytes(length: number = 1): Promise<Buffer> {
    return new Promise<Buffer>((resolve) => {
      this.reader.on('readable', () => {

        // 读取指定长度的字节
        let readedBuffer = this.reader.read(length)

        if (readedBuffer == null) {
          logd(`[readBytes] readedBuffer is null, stream buffer is empty`)
          return
        }

        // 可能返回string，转换成Buffer
        if (!Buffer.isBuffer(readedBuffer)) {
          readedBuffer = Buffer.from(readedBuffer)
        }
        // 移除监听器
        this.reader.removeAllListeners('readable')

        resolve(readedBuffer)
      })
    })
  }

stream

Nodejs

长度

字节

5 条回复 • 2023-07-26 16:01:12 +08:00

yaodong0126

2023-07-26 15:40:17 +08:00

感觉是你使用上出现了问题，既然使用了 stream ，一般来说是不需要自己控制数据的读取的，通过 pipe 处理就好了，即使使用三方库也是同样的逻辑

yaodong0126

2023-07-26 15:43:29 +08:00

举个例子
const { createReadStream, createWriteStream } = require('fs')

const readStream = createReadStream('/foo')
const writeStream = createWriteStream('/bar')
readStream.pipe(writeStream)

xlsepiphone

2023-07-26 15:48:03 +08:00

@yaodong0126 #2 我只是想按照字节读取，主要是解析二进制数据。看来还是只有自己创建缓冲区，多读几次。

yaodong0126

2023-07-26 15:55:46 +08:00

@xlsepiphone 如果需要这么做的，这属于 stream 的第二种使用方式，可以参考： https://nodejs.org/dist/latest-v20.x/docs/api/stream.html#two-reading-modes

xlsepiphone

2023-07-26 16:01:12 +08:00

更新了下代码，已结解决了:

```typescript
readBytes(length: number = 1): Buffer {
// 按照指定长度创建缓冲区
let buffer: Buffer = Buffer.alloc(length)
// 已经读取的字节长度
let readedLength = 0

for (;;) {
// 如果已经读取的字节长度等于指定的长度,则退出循环
if (readedLength === length) {
break
}
// 计算出可以读取的字节长度
const readableLength = Math.min(length - readedLength, this.reader.readableLength)
// 读取指定长度的字节
const readedBuffer = this.reader.read(readableLength)
// 填充到缓冲区中
buffer.fill(readedBuffer, readedLength)
// 更新已经读取的字节长度
readedLength += readableLength
}

return buffer
}
```