什么是JavaScript的异步生成器与for await...of循环，以及它们如何简化异步数据源的迭代操作？

async function*

for await...of

for await...of

for await...of

yield

await

async function* myAsyncGenerator() { /* ... */ yield await someAsyncOperation(); /* ... */ }

AsyncIterator

next()

{ value: ..., done: ... }

for await...of

next()

await

next()

await generator.next()

value

done

true

then()

catch()

for await...of

try...catch

async function fetchPages(url, startPage = 1, totalPages = 3) {
  let currentPage = startPage;
  while (currentPage <= totalPages) {
    console.log(`Fetching page ${currentPage}...`);
    try {
      const response = await fetch(`${url}?page=${currentPage}`);
      if (!response.ok) {
        throw new Error(`HTTP error! status: ${response.status}`);
      }
      const data = await response.json();
      yield data; // 每次yield一个页面的数据
      currentPage++;
    } catch (error) {
      console.error(`Error fetching page ${currentPage}:`, error.message);
      // 可以在这里选择是否继续，或者重新尝试
      // 为了演示，我们在这里直接抛出，让for await...of的try/catch捕获
      throw error; 
    }
    // 模拟网络延迟，让异步感更强
    await new Promise(resolve => setTimeout(resolve, 500)); 
  }
}

async function processAllPages() {
  const apiUrl = 'https://jsonplaceholder.typicode.com/posts'; // 这是一个模拟API，实际可能需要分页参数
  console.log("Starting to process all pages...");
  try {
    // 假设这个API实际支持分页，每次返回10个post
    // 这里为了演示，我们假设fetchPages能通过某种方式模拟分页数据
    for await (const pageData of fetchPages(apiUrl, 1, 3)) {
      console.log("Received page data:", pageData.slice(0, 2).map(p => p.title)); // 只打印前两个标题
      console.log(`Total items in this page: ${pageData.length}`);
    }
    console.log("Finished processing all pages.");
  } catch (error) {
    console.error("An error occurred during page processing:", error.message);
  }
}

// processAllPages(); // 实际运行时调用

fetchPages

yield

processAllPages

for await (const pageData of fetchPages(...))

yield

function*

function*

yield

next()

{ value: ..., done: ... }

value

async function*

yield

yield

async function*

await

next()

{ value: ..., done: ... }

for await...of

cursor

// 模拟一个异步API调用，它每次返回一部分日志和下一个游标
async function fetchLogBatch(cursor = null, limit = 5) {
  console.log(`API call: Fetching logs with cursor "${cursor || 'start'}", limit ${limit}...`);
  // 模拟网络延迟
  await new Promise(resolve => setTimeout(resolve, Math.random() * 1000 + 200)); 

  const allLogs = [
    { id: 1, timestamp: '2023-01-01T10:00:00Z', message: 'User logged in.' },
    { id: 2, timestamp: '2023-01-01T10:00:15Z', message: 'Data processed successfully.' },
    { id: 3, timestamp: '2023-01-01T10:00:30Z', message: 'Report generated.' },
    { id: 4, timestamp: '2023-01-01T10:00:45Z', message: 'User updated profile.' },
    { id: 5, timestamp: '2023-01-01T10:01:00Z', message: 'Payment initiated.' },
    { id: 6, timestamp: '2023-01-01T10:01:15Z', message: 'User logged out.' },
    { id: 7, timestamp: '2023-01-01T10:01:30Z', message: 'Background job started.' },
    { id: 8, timestamp: '2023-01-01T10:01:45Z', message: 'Cache cleared.' },
    { id: 9, timestamp: '2023-01-02T09:00:00Z', message: 'New feature deployed.' },
    { id: 10, timestamp: '2023-01-02T09:00:10Z', message: 'Database backup completed.' },
    { id: 11, timestamp: '2023-01-02T09:00:20Z', message: 'User registered.' },
    { id: 12, timestamp: '2023-01-02T09:00:30Z', message: 'Email sent.' },
  ];

  let startIndex = 0;
  if (cursor) {
    const cursorLog = allLogs.find(log => log.id === parseInt(cursor));
    if (cursorLog) {
      startIndex = allLogs.indexOf(cursorLog) + 1;
    }
  }

  const logs = allLogs.slice(startIndex, startIndex + limit);
  const nextCursor = logs.length > 0 ? logs[logs.length - 1].id.toString() : null;
  const hasMore = (startIndex + logs.length) < allLogs.length;

  return { logs, nextCursor, hasMore };
}

// 异步生成器：从日志服务获取所有日志
async function* getAllLogs(initialCursor = null) {
  let currentCursor = initialCursor;
  let hasMoreData = true;

  while (hasMoreData) {
    try {
      const { logs, nextCursor, hasMore } = await fetchLogBatch(currentCursor);

      if (logs.length === 0 && !hasMore) {
        // 没有更多数据了，并且当前批次为空，退出循环
        break; 
      }

      // 每次yield出获取到的日志批次
      yield logs; 

      currentCursor = nextCursor;
      hasMoreData = hasMore;

      if (!hasMoreData) {
        console.log("No more data to fetch. Generator will stop.");
      }
    } catch (error) {
      console.error("Error in getAllLogs generator:", error.message);
      // 如果发生错误，可以选择重新抛出，让消费者处理
      throw error; 
    }
  }
}

// 消费者：使用for await...of循环处理日志流
async function processLogStream() {
  console.log("Starting to process log stream...");
  let totalLogsProcessed = 0;
  try {
    for await (const logBatch of getAllLogs()) {
      console.log(`\n--- Processing a batch of ${logBatch.length} logs ---`);
      for (const log of logBatch) {
        console.log(`[${log.timestamp}] Log ID: ${log.id}, Message: "${log.message}"`);
        totalLogsProcessed++;
      }
      // 模拟处理每个批次可能需要一些时间
      await new Promise(resolve => setTimeout(resolve, 300)); 
    }
    console.log(`\nFinished processing log stream. Total logs processed: ${totalLogsProcessed}`);
  } catch (error) {
    console.error("An error occurred during log stream processing:", error.message);
  }
}

// 运行消费者
// processLogStream(); // 实际运行时调用

Promise.all

then()

for await...of

for await...of

for await...of