Python函数如何用生成器函数实现异步迭代 Python函数异步生成器的使用教程​

import asyncio

async def async_generator_function(data):
  """
  一个异步生成器函数，模拟异步数据流。
  """
  for item in data:
    await asyncio.sleep(1)  # 模拟I/O操作
    yield item

async def main():
  """
  主函数，使用async for循环迭代异步生成器。
  """
  data = [1, 2, 3, 4, 5]
  async for item in async_generator_function(data):
    print(f"处理: {item}")

if __name__ == "__main__":
  asyncio.run(main())

try...except

try

except

import asyncio

async def async_generator_with_error():
  """
  一个异步生成器函数，模拟可能发生异常的异步数据流。
  """
  for i in range(3):
    try:
      await asyncio.sleep(1)
      if i == 1:
        raise ValueError("模拟异常")
      yield i
    except ValueError as e:
      print(f"生成器内部捕获异常: {e}")
      yield None  # 或者抛出异常，取决于你的需求

async def main_with_error_handling():
  """
  主函数，使用try...except块处理异步生成器中的异常。
  """
  try:
    async for item in async_generator_with_error():
      print(f"处理: {item}")
  except ValueError as e:
    print(f"主函数捕获异常: {e}")

if __name__ == "__main__":
  asyncio.run(main_with_error_handling())

import asyncio

async def fetch_data_page(page_number, page_size):
  """
  模拟从数据源获取一页数据。
  """
  await asyncio.sleep(0.5)  # 模拟网络延迟
  # 假设数据源是一个列表
  data = [f"数据 {i}" for i in range((page_number - 1) * page_size, page_number * page_size)]
  return data

async def async_data_generator(page_size=10):
  """
  一个异步生成器函数，用于分页加载数据。
  """
  page_number = 1
  while True:
    data = await fetch_data_page(page_number, page_size)
    if not data:
      break  # 没有更多数据了
    for item in data:
      yield item
    page_number += 1

async def main_pagination():
  """
  主函数，使用async for循环迭代异步生成器，实现分页加载数据。
  """
  async for item in async_data_generator(page_size=5):
    print(f"处理: {item}")

if __name__ == "__main__":
  asyncio.run(main_pagination())

fetch_data_page

async_data_generator

page_size