JS `async Generator` 与 `for await…of`：异步迭代与流处理

各位观众，大家好！今天咱们来聊聊JavaScript里一对儿很有意思的组合：async Generator 和 for await...of。别害怕，名字听起来唬人，其实用起来挺简单，理解了之后你会发现它们在处理异步数据流的时候简直不要太方便。

先打个招呼：你瞅啥？瞅你咋地？瞅瞅这对儿黄金搭档！

好了，废话不多说，咱们直接进入正题。

一、什么是 Generator？（回顾一下基础）

在深入 async Generator 之前，我们先简单回顾一下普通的 Generator。Generator 函数是一种可以暂停执行，并在稍后恢复执行的函数。它和普通函数最大的区别在于：

*`function关键字：** 使用function*` 声明。
yield 关键字： 使用 yield 暂停执行，并返回一个值。

function* myGenerator() {
  yield 1;
  yield 2;
  yield 3;
}

const iterator = myGenerator();

console.log(iterator.next()); // { value: 1, done: false }
console.log(iterator.next()); // { value: 2, done: false }
console.log(iterator.next()); // { value: 3, done: false }
console.log(iterator.next()); // { value: undefined, done: true }

在这个例子中，myGenerator 函数就是一个 Generator。每次调用 iterator.next() 都会执行到下一个 yield 语句，并返回一个包含 value 和 done 属性的对象。value 是 yield 后的值，done 表示 Generator 是否已经执行完毕。

二、什么是 Async Generator？

Async Generator 就是 Generator 的异步版本。它允许我们在 Generator 函数中使用 await 关键字，从而处理异步操作。

*`async function关键字：** 使用async function*` 声明。
yield 关键字： yield 后面可以跟一个 Promise。

async function* myAsyncGenerator() {
  yield await Promise.resolve(1);
  yield await Promise.resolve(2);
  yield await Promise.resolve(3);
}

const asyncIterator = myAsyncGenerator();

asyncIterator.next().then(result => console.log(result)); // { value: 1, done: false }
asyncIterator.next().then(result => console.log(result)); // { value: 2, done: false }
asyncIterator.next().then(result => console.log(result)); // { value: 3, done: false }
asyncIterator.next().then(result => console.log(result)); // { value: undefined, done: true }

在这个例子中，myAsyncGenerator 函数就是一个 Async Generator。注意，每次调用 asyncIterator.next() 都会返回一个 Promise，我们需要使用 .then() 来获取结果。

三、什么是 for await...of？

for await...of 循环是专门用来迭代 Async Iterator 的。它会等待每个 Promise resolve 后，再执行循环体。

async function* myAsyncGenerator() {
  yield await Promise.resolve(1);
  yield await Promise.resolve(2);
  yield await Promise.resolve(3);
}

async function main() {
  for await (const value of myAsyncGenerator()) {
    console.log(value); // 1, 2, 3
  }
}

main();

在这个例子中，for await...of 循环会依次等待 Promise.resolve(1)，Promise.resolve(2)，Promise.resolve(3) resolve 后，分别将它们的值赋值给 value，并执行循环体。

四、为什么需要 Async Generator 和 for await...of？

传统的 Promise.all 和 async/await 在处理并发请求时很方便，但是它们需要一次性获取所有数据。如果我们需要处理一个无限的或者非常大的数据流，这种方式就不太合适了。Async Generator 和 for await...of 提供了一种按需获取数据的方式，可以有效地处理异步数据流。

五、 Async Generator 和 for await...of 的应用场景

读取大型文件： 我们可以使用 Async Generator 一部分一部分地读取大型文件，而不是一次性将整个文件加载到内存中。

const fs = require('fs');
const readline = require('readline');

async function* readFileByLines(filePath) {
  const fileStream = fs.createReadStream(filePath);
  const rl = readline.createInterface({
    input: fileStream,
    crlfDelay: Infinity
  });

  for await (const line of rl) {
    yield line;
  }
}

async function main() {
  for await (const line of readFileByLines('large_file.txt')) {
    console.log(line);
    // 对每一行进行处理
  }
}

main();

处理 WebSocket 数据流： 我们可以使用 Async Generator 接收 WebSocket 消息，并按需处理。

const WebSocket = require('ws');

async function* webSocketStream(url) {
  const ws = new WebSocket(url);

  yield new Promise((resolve, reject) => {
    ws.onopen = () => resolve(ws);
    ws.onerror = (error) => reject(error);
  });

  try {
    while (ws.readyState === WebSocket.OPEN) {
      yield new Promise((resolve, reject) => {
        ws.onmessage = (event) => resolve(event.data);
        ws.onerror = (error) => reject(error);
        ws.onclose = () => resolve(undefined); // Handle connection close
      });
    }
  } finally {
      ws.close();
  }

}

async function main() {
  try {
    for await (const message of webSocketStream('ws://localhost:8080')) {
      if (message === undefined) {
          console.log("WebSocket connection closed.");
          break;
      }
      console.log('Received:', message);
      // 对每一条消息进行处理
    }
  } catch (error) {
    console.error("WebSocket error:", error);
  }
}

main();

分页加载数据： 我们可以使用 Async Generator 从 API 分页加载数据，并按需展示。

async function* fetchPaginatedData(url, pageSize = 10) {
  let page = 1;
  let hasMore = true;

  while (hasMore) {
    const response = await fetch(`${url}?page=${page}&pageSize=${pageSize}`);
    const data = await response.json();

    if (data.length === 0) {
      hasMore = false;
    } else {
      yield* data; // 使用 yield* 迭代数组
      page++;
    }
  }
}

async function main() {
  for await (const item of fetchPaginatedData('https://api.example.com/data')) {
    console.log(item);
    // 对每一个数据项进行处理
  }
}

main();

六、 Async Generator 的高级用法

*`yield委托迭代：** 可以使用yield*` 将迭代委托给另一个 Generator 或者可迭代对象。在上面的分页加载数据例子中已经展示了用法。
return() 方法： Async Generator 实例具有 return() 方法，可以提前结束 Generator 的执行。

async function* myAsyncGenerator() {
  try {
    yield await Promise.resolve(1);
    yield await Promise.resolve(2);
    yield await Promise.resolve(3);
  } finally {
    console.log('Generator cleanup'); // 在 return() 调用时执行
  }
}

async function main() {
  const iterator = myAsyncGenerator();

  console.log(await iterator.next()); // { value: 1, done: false }
  console.log(await iterator.return('Early exit')); // { value: 'Early exit', done: true }
  console.log(await iterator.next()); // { value: undefined, done: true }
}

main();

throw() 方法： Async Generator 实例具有 throw() 方法，可以在 Generator 函数中抛出一个错误。

async function* myAsyncGenerator() {
  try {
    yield await Promise.resolve(1);
    yield await Promise.resolve(2);
    yield await Promise.resolve(3);
  } catch (error) {
    console.error('Generator caught error:', error);
  }
}

async function main() {
  const iterator = myAsyncGenerator();

  console.log(await iterator.next()); // { value: 1, done: false }
  try {
    await iterator.throw(new Error('Something went wrong'));
  } catch (error) {
    console.error('Main caught error:', error);
  }
  console.log(await iterator.next()); // { value: undefined, done: true }
}

main();

七、错误处理

在使用 for await...of 循环时，可以使用 try...catch 语句来捕获异步错误。

async function* myAsyncGenerator() {
  yield await Promise.resolve(1);
  throw new Error('Something went wrong');
  yield await Promise.resolve(2);
}

async function main() {
  try {
    for await (const value of myAsyncGenerator()) {
      console.log(value);
    }
  } catch (error) {
    console.error('Error in for await...of loop:', error);
  }
}

main();

八、与其他技术的比较

特性	Async Generator + `for await...of`	RxJS (Reactive Extensions for JavaScript)	Node.js Streams
数据处理方式	按需拉取 (Pull-based)	基于事件推送 (Push-based)	按需拉取
适用场景	处理大型数据流，按需获取数据	复杂的异步数据转换和组合	处理 I/O 数据流
学习曲线	相对简单	陡峭	中等
依赖	内置于 JavaScript	需要引入 RxJS 库	内置于 Node.js

九、总结

Async Generator 和 for await...of 是一对强大的组合，它们可以有效地处理异步数据流，提高代码的可读性和可维护性。它们特别适合于处理大型文件、WebSocket 数据流、分页加载数据等场景。

十、练习题

编写一个 Async Generator 函数，模拟一个无限滚动的列表，每次返回 10 条数据。
使用 for await...of 循环，从 Async Generator 函数中获取数据，并展示到页面上。
添加错误处理机制，处理 Async Generator 函数中可能出现的错误。

十一、最后的温馨提示

Async Generator 和 for await...of 需要 ES2018 (ES9) 及以上版本的 JavaScript 环境支持。
在实际开发中，需要根据具体场景选择合适的技术方案。
多练习，多思考，才能真正掌握这些技术。

好了，今天的讲座就到这里。希望大家有所收获！咱们下次再见！

发表回复 取消回复

发表回复取消回复