Python aiohttp / asyncio - 如何处理返回的数据

2 回答

• 7

  这是concurrent.futures.ProcessPoolExecutor的示例 . 如果在未指定 max_workers 的情况下创建它，则实现将使用 os.cpu_count . 另请注意，asyncio.wrap_future是公开的，但没有记录 . 或者，有AbstractEventLoop.run_in_executor .

  import asyncio
  from concurrent.futures import ProcessPoolExecutor
  
  import aiohttp
  import lxml.html
  
  
  def process_page(html):
      '''Meant for CPU-bound workload'''
      tree = lxml.html.fromstring(html)
      return tree.find('.//title').text
  
  
  async def fetch_page(url, session):
      '''Meant for IO-bound workload'''
      async with session.get(url, timeout = 15) as res:
        return await res.text()
  
  
  async def process(url, session, pool):
      html = await fetch_page(url, session)
      return await asyncio.wrap_future(pool.submit(process_page, html))
  
  
  async def dispatch(urls):
      pool = ProcessPoolExecutor()
      async with aiohttp.ClientSession() as session:
          coros = (process(url, session, pool) for url in urls)
          return await asyncio.gather(*coros)
  
  
  def main():
      urls = [
        'https://stackoverflow.com/',
        'https://serverfault.com/',
        'https://askubuntu.com/',
        'https://unix.stackexchange.com/'
      ]
      result = asyncio.get_event_loop().run_until_complete(dispatch(urls))
      print(result)
  
  if __name__ == '__main__':
      main()
  

  回复于 2024-05-01T03:35:07+08:00
• 5

  您的代码离标记不远 . asyncio.gather 按参数的顺序返回结果，因此在此处保留顺序，但不会按顺序调用 page_content .

  一些调整：

  首先，你不需要 ensure_future . 只有当你试图使一个协程比其父节点更长时才需要创建一个任务，即如果任务必须继续运行，即使创建它的函数已经完成 . 这里你需要的是直接用你的协同程序调用 asyncio.gather ：

  async def get_url_data(urls, username, password):
      async with aiohttp.ClientSession(...) as session:
          responses = await asyncio.gather(*(fetch(session, i) for i in urls))
      for i in responses:
          print(i.title.text)
      return responses
  

  But 调用它会同时安排所有提取，并且有大量的URL，这远非最佳 . 相反，您应该选择最大并发性，并确保最多X次提取随时都在运行 . 要实现这一点，你可以使用 asyncio.Semaphore(20) ，这个信号量只能被最多20个协同程序获取，所以其他人将等待获取直到有一个地点可用 .

  CONCURRENCY = 20
  TIMEOUT = 15
  
  async def fetch(session, sem, url):
      async with sem:
          async with session.get(url) as response:
              return page_content(await response.text())
  
  async def get_url_data(urls, username, password):
      sem = asyncio.Semaphore(CONCURRENCY)
      async with aiohttp.ClientSession(...) as session:
          responses = await asyncio.gather(*(
              asyncio.wait_for(fetch(session, sem, i), TIMEOUT)
              for i in urls
          ))
      for i in responses:
          print(i.title.text)
      return responses
  

  这样，所有提取都会立即启动，但只有20个能够获取信号量 . 其他人将在第一个 async with 指令处阻塞，并等待另一次提取完成 .

  我也用这里的官方asyncio替换了aiohttp.Timeout .

  最后，对于数据的实际处理，如果受到CPU时间的限制，asyncio可能对您没有多大帮助 . 您需要在此处使用 ProcessPoolExecutor 将实际工作与另一个CPU并行化 . run_in_executor 可能会有用 .

  回复于 2024-05-01T03:35:07+08:00