跳转到正文

流

以 Markdown 格式查看 在 GitHub 上编辑 文档位置 5b53f69

Phasis实现了完整的 WHATWG Streams Standard。每个类、每种读取器模式、每个辅助方法一应俱全。在 Node 和现代浏览器中可运行的相同代码，也能在 Phasis 中运行。

这正是 fetch内部使用的实现 — Response.body是真正的 ReadableStream，Request接受流式请求体，而且克隆 Response 所需的 tee()也能正常工作。

已发布内容

类 状态
ReadableStream 完整
ReadableStreamDefaultController 完整
ReadableByteStreamController 完整
ReadableStreamDefaultReader 完整
ReadableStreamBYOBReader 完整
ReadableStreamBYOBRequest 完整
WritableStream 完整
WritableStreamDefaultController 完整
WritableStreamDefaultWriter 完整
TransformStream 完整
TransformStreamDefaultController 完整
ByteLengthQueuingStrategy 完整
CountQueuingStrategy 完整

以及常用的方法：pipeTo, pipeThrough, tee, getReader、异步迭代、Symbol.asyncIterator，以及配合 { min } 使用的 BYOB 读取。

生成

JavaScript
// Pull-based — only generates values when read
const counter = new ReadableStream({
  start(controller) { this.n = 0; },
  pull(controller) {
    if (this.n >= 5) { controller.close(); return; }
    controller.enqueue(this.n++);
  },
});

for await (const v of counter) console.log(v);  // 0,1,2,3,4
JavaScript
// Push-based — start enqueues everything eagerly
const small = new ReadableStream({
  start(c) {
    [10, 20, 30].forEach(n => c.enqueue(n));
    c.close();
  },
});

消费

三种读取方式：

JavaScript
// 1. Reader
const reader = stream.getReader();
while (true) {
  const { value, done } = await reader.read();
  if (done) break;
  console.log(value);
}
reader.releaseLock();

// 2. Async iteration (preferred)
for await (const chunk of stream) {
  console.log(chunk);
}

// 3. tee + parallel pipes
const [a, b] = stream.tee();
await Promise.all([a.pipeTo(sinkA), b.pipeTo(sinkB)]);

转换

JavaScript
const upper = new TransformStream({
  transform(chunk, controller) {
    controller.enqueue(chunk.toUpperCase());
  },
});

const result = new ReadableStream({
  start(c) { ['hi', 'there'].forEach(s => c.enqueue(s)); c.close(); }
}).pipeThrough(upper);

for await (const v of result) console.log(v);  // HI, THERE

字节流（BYOB）

JavaScript
const bytes = new ReadableStream({
  type: 'bytes',
  start(controller) {
    controller.enqueue(new Uint8Array([1, 2, 3, 4, 5, 6]));
    controller.close();
  },
});

const reader = bytes.getReader({ mode: 'byob' });
const buf = new Uint8Array(4);
const { value, done } = await reader.read(buf);
// value is a Uint8Array of length 4, sharing buffer with `buf`

{ min: N }受支持 — 只有至少有 2 个元素可用时，read(view, { min: 2 }) 才会完成解析。

背压

desiredSize和 ready 的工作方式符合规范。可使用排队策略为生产者提供反向通道：

JavaScript
const stream = new ReadableStream({
  pull(controller) {
    if (controller.desiredSize <= 0) return;  // back off
    controller.enqueue(nextChunk());
  },
}, new CountQueuingStrategy({ highWaterMark: 4 }));

与 fetch 配合使用

JavaScript
// Response body is a ReadableStream
const res = await fetch('https://example.com/big.txt');
let total = 0;
for await (const chunk of res.body) {
  total += chunk.byteLength;
}

// Request bodies can stream too
const upload = new ReadableStream({
  start(c) { c.enqueue(new TextEncoder().encode('hello')); c.close(); }
});
await fetch('https://example.com/upload', { method: 'POST', body: upload });

使用 tee 缓存

常见模式：读取请求体两次。

JavaScript
const res = await fetch(url);
const [forCache, forUse] = res.body.tee();
await Promise.all([
  forCache.pipeTo(cacheWritable),
  parseJson(forUse),
]);

Response.clone()会为你完成此操作，并返回一个新的 Response。

注意事项

  • 仅支持同步操作的生产者。 一个 pull() 执行阻塞式 PHP I/O 时会阻塞脚本。对于 HTTP，请使用 fetch 传输钩子；对于文件，优先预先读入内存。setTimeout / setInterval 可用（参见 事件循环），但引擎默认使用虚拟时间——在 pull() 内的 setTimeout(cb, 1000) 不会休眠 PHP，而是推进循环时钟。要按真实时间调度，需要宿主根据墙上时钟驱动 tickEventLoop()。

另请参阅