const reader = stream.getReader();
let bytesReceived = 0;
const processData = (result) => {
    if (result.done) {
        console.log(`complete, total size: ${bytesReceived}`);
        return;
    const value = result.value; // Uint8Array
    const length = value.length;
    console.log(`got ${length} bytes data:`, value);
    bytesReceived += length;
    // 读取下一个文件片段，重复处理步骤
    return reader.read().then(processData);
reader.read().then(processData);​

其中 result.value 参数为这次读取得到的片段，它是一个 Uint8Array，通过循环调用 reader.read() 方法就能一点点地获取流的整个数据；而 result.done 参数负责表明这个流是否已经读取完毕，当 result.done 为 true 时表明流已经关闭，不会再有新的数据，此时 result.value 的值为 undefined。

回到我们之前的问题，我们可以通过读取 Response 中的流得到正在接收的文件片段，累加各个片段的 length 就能得到类似 XHR onprogress 事件的 loaded，也就是已下载的字节数；通过从 Response 的 headers 中取出 Content-Length 就能得到类似 XHR onprogress 事件的 total，也就是总字节数。于是我们可以写出下面的代码，成功得到下载进度：

let total = null;
let loaded = 0;
const logProgress = (reader) => {
    return reader.read().then(({ value, done }) => {
        if (done) {
            console.log('Download completed');
            return;
        loaded += value.length;
        if (total === null) {
            console.log(`Downloaded ${loaded}`);
        } else {
            console.log(`Downloaded ${loaded} of ${total} (${(loaded / total * 100).toFixed(2)}%)`);
        return logProgress(reader);
fetch('/foo').then((res) => {
    total = res.headers.get('content-length');
    return res.body.getReader();
}).then(logProgress);

看着好像没问题是吧？问题来了，实际运行后，数据呢？我那么大一个返回数据呢？上面的代码只顾着输出进度了，结果并没有把返回数据传回来。虽然我们可以直接在上面的代码里处理二进制数据片段，可是有时我们还是会偷懒，直接得到完整的数据进行处理（比如一个巨大的 JSON 字符串）。

如果我们希望接收的数据是文本，一种解决方案是借助 TextDecoder 得到解析后的文本并拼接，最后将整个文本返回：

let text = '';
const logProcess = (res) => {
    const reader = res.body.getReader();
    const decoder = new TextDecoder('utf-8');
    const push = ({ value, done }) => {
        if (done) return JSON.parse(text);
        text += decoder.decode(value, { stream: true });
        // ...
        return reader.read().then(push);
    return reader.read().then(push);
fetch('/foo').then(logProgress).then((res) => { ... });

不过如果你犯了强迫症，一定要像原来那样显示调用 res.json() 之类的方法得到数据，这该怎么办呢？既然 fetch() 方法返回一个 Response 对象，而这个对象的数据已经在 ReadableStream 中读取下载进度时被使用了，那我再构造一个 ReadableStream，外面再包一个 Response 对象并返回，问题不就解决了吗？

构造一个 ReadableStream

构造一个 ReadableStream 时可以定义以下方法和参数：

const stream = new ReadableStream({
    start(controller) {
        // start 方法会在实例创建时立刻执行，并传入一个流控制器
        controller.desiredSize
            // 填满队列所需字节数
        controller.close()
            // 关闭当前流
        controller.enqueue(chunk)
            // 将片段传入流的队列
        controller.error(reason)
            // 对流触发一个错误
    pull(controller) {
        // 将会在流的队列没有满载时重复调用，直至其达到高水位线
    cancel(reason) {
        // 将会在流将被取消时调用
}, queuingStrategy); // { highWaterMark: 1 }

而构造一个 Response 对象就简单了，Response 对象的第一个参数即是返回值，可以是字符串、Blob、TypedArray，甚至是一个 Stream；而它的第二个参数则和 fetch() 方法很像，也是一些初始化参数。

const response = new Response(source, init);

了解以上的内容后，我们只需要构造一个 ReadableStream，然后把「从 reader 中循环读取数据」的逻辑放在这个流的 start() 方法内，它会在流实例化后立即调用。当 reader 读取数据时可以输出下载进度，同时调用 controller.enqueue() 把得到的数据推进我们构造出来的流，最后在读取完毕时调用 controller.close() 关闭这个流，问题就能轻松解决。

const logProgress = (res) => {
    const total = res.headers.get('content-length');
    let loaded = 0;
    const reader = res.body.getReader();
    const stream = new ReadableStream({
        start(controller) {
            const push = () => {
                reader.read().then(({ value, done }) => {
                    if (done) {
                        controller.close();
                        return;
                    loaded += value.length;
                    if (total === null) {
                        console.log(`Downloaded ${loaded}`);
                    } else {
                        console.log(`Downloaded ${loaded} of ${total} (${(loaded / total * 100).toFixed(2)}%)`);
                    controller.enqueue(value);
                    push();
            push();
    return new Response(stream, { headers: res.headers });
fetch('/foo').then(logProgress).then(res => res.json()).then((data) => { ... });

分流一个 ReadableStream

感觉是不是绕了一个远路？就为了这点功能我们居然构造了一个 ReadableStream 实例？有没有更简单的方法？其实是有的，如果你稍有留意的话，应该会注意到 ReadableStream 实例上有一个名字看起来有点奇怪的 tee() 方法。这个方法可以将一个流分流成两个一模一样的流，两个流可以读取完全相同的数据。

所以我们可以利用这个特性将一个流分成两个流，将其中一个流用于输出下载进度，而另一个流直接返回：

const logProgress = (res) => {
    const total = res.headers.get('content-length');
    let loaded = 0;
    const [progressStream, returnStream] = res.body.tee();
    const reader = progressStream.getReader();
    const log = () => {
        reader.read().then(({ value, done }) => {
            if (done) return;
            // 省略输出进度
            log();
    log();
    return new Response(returnStream, { headers: res.headers });
fetch('/foo').then(logProgress).then(res => res.json()).then((data) => { ... });

另外其实 fetch 请求返回的 Response 实例上有一个一看就知道是什么意思的 clone() 方法，这个方法可以得到一个克隆的 Response 实例。所以我们可以将其中一个实例用来获取流并得到下载进度，另一个实例直接返回，这样就省去了构造 Response 的步骤，效果是一样的。其实这个方法一般用在 Service Worker 里，例如将请求得到的结果缓存起来等等。

很好，下载进度的问题完美解决了，那么让我们回到最早的问题。Fetch API 最早是没有 signal 这个参数的，所以早期的 fetch 请求很难中断——对，是「很难」，而不是「不可能」。如果浏览器实现了 ReadableStream 并在 Response 上提供了 body 的话，是可以通过流的中断实现这个功能的。

中断一个 ReadableStream

总结一下我们现在已经知道的内容，fetch 请求返回一个 Response 对象，从中可以得到一个 ReadableStream，然后我们还知道了如何自己构造 ReadableStream 和 Response 对象。再回过头看看 ReadableStream 实例上还没提到的方法，想必你一定注意到了那个 cancel() 方法。

通过 ReadableStream 上的 cancel() 方法，我们可以关闭这个流。此外你可能也注意到 reader 上也有一个 cancel() 方法，这个方法的作用是关闭与这个 reader 相关联的流，所以从结果上来看，两者是一样的。而对于 Fetch API 来说，关闭返回的 Response 对象的流的结果就相当于中断了这个请求。

所以，我们可以像之前那样构造一个 ReadableStream 用于传递从 res.body.getReader() 中得到的数据，并对外暴露一个 aborter() 方法。调用这个 aborter() 方法时会调用 reader.cancel() 关闭 fetch 请求返回的流，然后调用 controller.error() 抛出错误，中断构造出来的传递给后续操作的流：

let aborter = null;
const abortHandler = (res) => {
    const reader = res.body.getReader();
    const stream = new ReadableStream({
        start(controller) {
            let aborted = false;
            const push = () => {
                reader.read().then(({ value, done }) => {
                    if (done) {
                        if (!aborted) controller.close();
                        return;
                    controller.enqueue(value);
                    push();
            aborter = () => {
                reader.cancel();
                controller.error(new Error('Fetch aborted'));
                aborted = true;
            push();
    return new Response(stream, { headers: res.headers });
fetch('/foo').then(abortHandler).then(res => res.json()).then((data) => { ... });
aborter();

流的锁机制

或许你还是很奇怪，既然流本身就有一个 cancel() 方法，为什么我们不直接暴露这个方法，反而要绕路构造一个新的 ReadableStream 呢？例如像下面这样：

let aborter = null;
const abortHandler = (res) => {
    aborter = () => res.body.cancel();
    return res;
fetch('/foo').then(abortHandler).then(res => res.json()).then((data) => { ... });
aborter();

可惜这样执行会得到下面的错误：这个流被锁了。

TypeError: Failed to execute 'cancel' on 'ReadableStream': Cannot cancel a locked stream

你不信邪，既然流的 reader 被关闭时会关闭相关联的流，那么只要再获取一个 reader 并 cancel() 不就好了？

let aborter = null;
const abortHandler = (res) => {
    aborter = () => res.body.getReader().cancel();
    return res;
fetch('/foo').then(abortHandler).then(res => res.json()).then((data) => { ... });
aborter();

可惜这样执行还是会得到下面的错误：

TypeError: Failed to execute 'getReader' on 'ReadableStream': ReadableStreamReader constructor can only accept readable streams that are not yet locked to a reader

或许你还会想，像之前那样使用 tee() 克隆一个流，然后关闭克隆的流不就好了？可惜即便成功调用了其中一个流的 cancel() 方法，请求还是没有中断，因为另一个流并没有被中断，并且还在不断地接收数据。

于是我们接触到了流的锁机制。一个流只能同时有一个处于活动状态的 reader，当一个流被一个 reader 使用时，这个流就被该 reader 锁定了，此时流的 locked 属性为 true。如果这个流需要被另一个 reader 读取，那么当前处于活动状态的 reader 可以调用 reader.releaseLock() 方法释放锁。此外 reader 的 closed 属性是一个 Promise，当 reader 被关闭或者释放锁时，这个 Promise 会被 resolve，可以在这里编写关闭 reader 的处理逻辑：

reader.closed.then(() => {
  console.log('reader closed');
reader.releaseLock();

可是上面的代码似乎没用上 reader 啊？再仔细思考下 res => res.json() 这段代码，是不是有什么启发？

让我们翻一下 Fetch API 的规范文档，在 5.2. Body mixin 中 有如下一段话：

Objects implementing the Body mixin also have an associated consume body algorithm, given a _type_, runs these steps:

If this object is disturbed or locked, return a new promise rejected with a TypeError.
Let stream be body’s stream if body is non-null, or an empty ReadableStream object otherwise.
Let reader be the result of getting a reader from _stream_. If that threw an exception, return a new promise rejected with that exception.
Let promise be the result of reading all bytes from stream with _reader_.
Return the result of transforming promise by a fulfillment handler that returns the result of the package data algorithm with its first argument, type and this object’s MIME type.

简单来说，当我们调用 Body 上的方法时，浏览器隐式地创建了一个 reader 读取了返回数据的流，并创建了一个 Promise 实例，待所有数据被读取完后再 resolve 并返回格式化后的数据。所以，当我们调用了 Body 上的方法时，其实就创建了一个我们无法接触到的 reader，此时这个流就被锁住了，自然也无法从外部取消。

最后，我们看个实例

示例：断点续传

现在我们可以随时中断一个请求，以及获取到请求过程中的数据，甚至还能修改这些数据。或许我们可以用来做些有趣的事情，比如各个下载器中非常流行的断点续传功能。

首先我们先来了解下断点续传的原理，简述如下：