20 | 揭秘 Python 协程

精选留言(83)

• 

  Jingxiao 置顶

  发现评论区好多朋友说无法运行，在这里统一解释下：
  1. %time 是 jupyter notebook 自带的语法糖，用来统计一行命令的运行时间；如果你的运行时是纯粹的命令行 python，或者 pycharm，那么请把 %time 删掉，自己用传统的时间戳方法来记录时间也可以；或者使用 jupyter notebook
  2. 我的本地解释器是 Anaconda Python 3.7.3，亲测 windows / ubuntu 均可正常运行，如无法执行可以试试 pip install nest-asyncio，依然无法解决请尝试安装 Anaconda Python
  3. 这次代码因为使用了较新的 API，所以需要较新的版本号，但是朋友们依然出现了一些运行时问题，这里先表示下歉意；同时也想说明的是，在提问之前自己经过充分搜索，尝试后解决问题，带来的快感，和能力的提升，相应也是很大的，一门工程最需要的是 hands on dirty work（动手做脏活），才能让自己的能力得到本质的提升，加油！

  2019-06-25

   4

   34
• 

  Jingxiao 置顶

  思考题答案：
  
  在 python 3.7 及以上的版本中，我们对 task 对象调用 add_done_callback() 函数，即可绑定特定回调函数。回调函数接受一个 future 对象，可以通过 future.result() 来获取协程函数的返回值。
  
  示例如下：
  
  import asyncio
  
  async def crawl_page(url):
      print('crawling {}'.format(url))
      sleep_time = int(url.split('_')[-1])
      await asyncio.sleep(sleep_time)
      return 'OK {}'.format(url)
  
  async def main(urls):
      tasks = [asyncio.create_task(crawl_page(url)) for url in urls]
      for task in tasks:
          task.add_done_callback(lambda future: print('result: ', future.result()))
      await asyncio.gather(*tasks)
  
  %time asyncio.run(main(['url_1', 'url_2', 'url_3', 'url_4']))
  
  输出：
  
  crawling url_1
  crawling url_2
  crawling url_3
  crawling url_4
  result: OK url_1
  result: OK url_2
  result: OK url_3
  result: OK url_4
  Wall time: 4 s

  2019-07-01

  

   9
• 

  Wing·三金

  思考题：简单的理解是每个新建的协程对象都会自动调用 add_done_callback() 函数来添加一个回调函数，当协程对象的 Future 状态启动时就会调用该回调函数，从而实现回调。
  
  综合下前面的留言和个人的学习，总结下 py 3.6 版本下 asyncio 的主要不同：
  1、没有 run(), create_task()，可以用 asyncio.get_even_loop().run_until_complete() 来代替 run()，用 ensure_future() 来代替 create_task()；
  2、可能会出现 RuntimeError: This event loop is already running，解决方案一：pip install nest_asyncio; import nest_asyncio; nest_asyncio.apply()；解决方案二：有些友人说是 tornado 5.x 起的版本才有此问题，可考虑将其版本降至 4.x（不推荐）；
  3、%time 与 %%time 的主要区别：%time func()（必须是同一行）；%%time 必须放在单元格的开头，强烈建议单独一行 + 不要与 import、def 相关的语句放在同个单元格；
  4、爬虫中的 aiohttp.ClientSession(headers=header, connector=aiohttp.TCPConnector(ssl=False)) 提及未声明的 header，要么将 headers=header 部分去掉使用默认参数，要么用诸如 header={"User-Agent": "Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/74.0.3729.157 Safari/537.36"} 来显式声明；
  5、tasks = [asyncio.create_task(crawl_page(url)) for url in urls]; await asyncio.gather(*tasks);
  约等于
  tasks = [crawl_page(url) for url in urls]; asyncio.get_even_loop().run_until_complete(asyncio.wait(tasks));
  或
  tasks = [asyncio.ensure_future(crawl_page(url)) for url in urls]; await asyncio.gather(*tasks);

  2019-06-27

  

   15
• 

  cuikt

  在代码async with aiohttp.ClientSession(headers=header, connector=aiohttp.TCPConnector(ssl=False)) as session:中的headers=header需要定义header为一个字典可以像下面这样，否则会报错。
  header={"User-Agent": "Mozilla/5.0 (Windows NT 6.1; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/74.0.3729.157 Safari/537.36"}

  2019-06-25

  

   13
• 

  helloworld

  说一下我对await的理解：
  开发者要提前知道一个任务的哪个环节会造成I/O阻塞，然后把这个环节的代码异步化处理，并且通过await来标识在任务的该环节中断该任务执行，从而去执行下一个事件循环任务。这样可以充分利用CPU资源，避免CPU等待I/O造成CPU资源白白浪费。当之前任务的那个环节的I/O完成后，线程可以从await获取返回值，然后继续执行没有完成的剩余代码。
  由上面分析可知，如果一个任务不涉及到网络或磁盘I/O这种耗时的操作，而只有CPU计算和内存I/O的操作时，协程并发的性能还不如单线程loop循环的性能高。

  2019-06-27

   3

   12
• 

  rogerr

  目前看到进阶篇就很痛苦了，对于小白来说，基础篇还好理解，到这儿就对深层次的知识无法深入理解和实践，感觉渐行渐远，不知道如何深入下去

  2019-06-25

   5

   10
• 

  tt

  学习笔记:异步和阻塞。
  
  阻塞主要是同步编程中的概念:执行一个系统调用，如果暂时没有返回结果，这个调用就不会返回，那这个系统调用后面的应用代码也不会执行，整个应用被“阻塞”了。
  
  同步编程也可以有非阻塞的方式，在系统调用没有完成时直接返回一个错误码。
  
  异步调用是系统调用完成后返回应用一个消息，应用响应消息，获取结果。
  
  上面说的系统调用 对应Python中的异步函数——一个需要执行较长时间的任务。响应消息应该对应回调函数，但我觉得await异步函数返回本身就相当于系统告知应用系统调用完成了，后面的代码起码可以完成部分回调函数做的事情。
  
  阻塞和异步必须配合才能完成异步编程。
  
  1、await异步函数会造成阻塞;
  2、await一个task不会造成阻塞，但是task对应的异步函数中必定几乎await一个异步函数，这个异步函数会阻塞这个task的执行，这样其余task才能被事件调度器的调度从而获取执行机会。

  2019-06-24

   2

   5
• 

  Fergus

  # -*- coding:utf-8 -*-
  '''
  sublime
  py -3.6
  asyncio.gather(*args) 并发
  '''
  
  
  import time
  import asyncio
  
  async def crawl_page(url):
      print('crawling {}'.format(url))
      sleep_time = int(url.split('_')[-1])
      await asyncio.sleep(sleep_time)
      print('OK {}'.format(url))
      
  star = time.perf_counter()
  loop = asyncio.get_event_loop()
  tasks = (crawl_page(url) for url in ['url_1', 'url_2', 'url_3', 'url_4'])
  loop.run_until_complete(asyncio.gather(*tasks))
  loop.close()
  
  print('Wall time: {:.2f}'.format(time.perf_counter() - star))
  
  
  crawling url_4
  crawling url_1
  crawling url_2
  crawling url_3
  OK url_1
  OK url_2
  OK url_3
  OK url_4
  Wall time: 4.00
  [Finished in 4.5s]

  2019-06-27

   1

   4
• 

  Fergus

  # -*- encoding: utf-8 -*-
  '''
  aiohttp + aysncio爬取电豆瓣电影
  py 3.6
  sublime text3
  '''
  
  
  import time
  import aiohttp
  import asyncio
  from bs4 import BeautifulSoup
  
  now = lambda: time.perf_counter()
  
  async def fetchHtmlText(url):
      async with aiohttp.ClientSession(
          headers={'users-agent':'Mozilla/5.0'},
          connector=aiohttp.TCPConnector(ssl=False)) as session:
          async with session.get(url) as response:
              return await response.text()
  
  async def main():
      url = "https://movie.douban.com/cinema/later/beijing/"
      html = await fetchHtmlText(url)
      soup = BeautifulSoup(html, "html.parser")
  
      divs = soup.find_all('div', class_='item mod')
      urls = list(map(lambda x: x.a.img['src'], divs))
      names = list(map(lambda x: x.h3.a.string, divs))
      dats = list(map(lambda x: x.ul.li.string, divs))
  
      lis = zip(names, dats, urls)
      for i in lis:
          print("{0:{3}^25} \t {1:{3}^10} \t {2:{3}^}".format(i[0], i[1], i[2],chr(12288)))
  
  
  start = now()
  loop = asyncio.get_event_loop()
  loop.run_until_complete(main())
  print("Wall time: {}".format(now() - start))
  
  
  # 九龙不败　　　　　　　　　　　　　　　　　　　　　 　　07月02日　　 https://img3.doubanio.com/view/photo/s_ratio_poster/public/p2560169035.jpg
  # 别岁　　　　　　　　　　　　　　　　　　　　　　　 　　07月02日　　 https://img3.doubanio.com/view/photo/s_ratio_poster/public/p2558138041.jpg
  ...
  # 刀背藏身　　　　　　　　　　　　　　　　　　　　　 　　07月19日　　 https://img1.doubanio.com/view/photo/s_ratio_poster/public/p2557644589.jpg
  # Wall time: 1.994356926937086
  # [Finished in 3.7s]
  

  2019-06-28

  

   3
• 

  胡峣

  对于熟悉线程模型的程序员来讲，如果不把时间循环解释清楚，协程确实很难理解。
  我的理解协程在编程语言层面实现aio，只有io等待才切换上下文而不像线程按scheduler分片cpu在没必要的时候也切换上下文。不知道我理解的对不对。

  2019-07-06

  

   2
• 

  Fergus

  # -*- encoding: utf-8 -*-
  '''
  py3.6
  sublime
  '''
  
  
  import time
  import asyncio
  
  now = lambda: time.perf_counter()
  
  async def work_1():
  await asyncio.sleep(1)
  return 1
  
  async def work_2():
  await asyncio.sleep(2)
  return 2
  
  async def work_3():
  await asyncio.sleep(3)
  return 3
  
  async def main():
  
  task_1 = asyncio.ensure_future(work_1())
  task_2 = asyncio.ensure_future(work_2())
  task_3 = asyncio.ensure_future(work_3())
  
  await asyncio.sleep(2)
  task_3.cancel()
  
  ret = await asyncio.gather(*(task_1, task_2, task_3),
  return_exceptions=True)
  print(ret)
  
  star = now()
  loop = asyncio.get_event_loop()
  loop.run_until_complete(main())
  loop.close()
  print('Wall time:', now() - star)
  
  # [1, 2, CancelledError()]
  # Wall time: 2.0061719078276914
  # [Finished in 2.5s]

  2019-06-27

  

   2
• 

  Fergus

  # 例1 3.7以前的版本实现
  import time
  import asyncio
  
  # < py 3.7
  async def crawl_page(url):
      print('crawling {}'.format(url))
      sleep_time = int(url.split('_')[-1])
      r = await asyncio.sleep(sleep_time)
      print('OK {}'.format(url))
  
  async def main(urls):
      for url in urls:
          await crawl_page(url)
          
  star = time.perf_counter()
  loop = asyncio.get_event_loop()
  loop.run_until_complete(main(['url_1', 'url_2', 'url_3', 'url_4']))
  loop.close()
  # await main(['url_1', 'url_2', 'url_3', 'url_4'])
  print(time.perf_counter() - star)
  
  
  crawling url_1
  OK url_1
  crawling url_2
  OK url_2
  crawling url_3
  OK url_3
  crawling url_4
  OK url_4
  10.003888006104894

  2019-06-26

  

   2
• 

  Fergus

  # 例一修改 threading
  
  import time
  import threading
  
  def crawl_page(url):
      print('crawling {}'.format(url))
      sleep_time = int(url.split('_')[-1])
      time.sleep(sleep_time)
      print('OK {}'.format(url))
      print('耗时:{:.2f}'.format(time.perf_counter() - star))
      
  def main(urls):
      tasks = [threading.Thread(target=crawl_page, name=url, args=(url,)) for url in urls]
      for t in tasks:
          t.start()
          print(threading.current_thread().name)
      
  star = time.perf_counter()
  main(['url_1', 'url_2', 'url_3', 'url_4'])
  
  crawling url_1MainThread
  crawling url_2MainThread
  crawling url_3MainThread
  crawling url_4MainThread
  OK url_1
  耗时:1.03
  OK url_2
  耗时:2.06
  OK url_3
  耗时:3.08
  OK url_4
  耗时:4.11

  2019-06-26

  

   2
• 

  刘朋

  当在 jupyter 中运行: %time asyncio.run(main(['url_1', 'url_2', 'url_3', 'url_4']))
  出现报错: RuntimeError: asyncio.run() cannot be called from a running event loop
  原因是: The problem in your case is that jupyter (IPython) is already running an event loop (for IPython ≥ 7.0)
  解决是: 将 %time asyncio.run(main(['url_1', 'url_2', 'url_3', 'url_4'])) 换成 await main(['url_1', 'url_2', 'url_3', 'url_4'])

  2019-06-26

   2

   2
• 

  毛利

  asyncio.run() cannot be called from a running event loop

  2019-10-31

  

   1
• 

  赤城

  翻阅评论，发现大家也有相似的问题，就是在老师的第二个代码块中，代码：
  %time asyncio.run(main(['url_1', 'url_2', 'url_3', 'url_4']))
  这行代码因为前面有%time语法糖，所以大家想当然就在jupyter中执行了，但是很多人都遇到了：
  RuntimeError: asyncio.run() cannot be called from a running event loop
  查阅Stack Overflow的解释jupyter已经一个event loop（https://stackoverflow.com/questions/55409641/asyncio-run-cannot-be-called-from-a-running-event-loop），因此需要将代码改成：
  await main(['url_1', 'url_2', 'url_3', 'url_4'])，而且不能在await前面加%time，否则会报错：
  SyntaxError: 'await' outside function
  新手学习协程感觉各种懵（苦笑脸）

  2019-09-23

  

   1
• 

  xavier

  老师你好，请教个问题。如下：
  
  文中代码：
  
  async def main():
      task1 = asyncio.create_task(worker_1())
      task2 = asyncio.create_task(worker_2())
      print('before await')
      await task1
      print('awaited worker_1')
      await task2
      print('awaited worker_2')
  
  请问调度器从何时开始启动调度？是一有任务创建就执行调度还是执行await task1时？老师这里说的是执行await task1，事件调度器开始调度 worker_1。但后面的生产者消费者模型，没有await操作，任务已开始运行。

  2019-07-19

  

   1
• 

  某彬

  关于以下一段代码：
  ```python
  for task in tasks:
   await task
  ```
  我想问 遇到第一个await之后程序不会跳出整个for循环吗？还是说只会跳出当前的循环？

  2019-07-16

  

   1
• 

  converse✪

  #coding:utf-8
  import asyncio
  import time
  
  async def corou1():
      print("corou 1, {}".format(time.perf_counter()))
      await asyncio.sleep(2)
      print("corou 1, {}".format(time.perf_counter()))
      
  async def corou2():
      print("corou 2, {}".format(time.perf_counter()))
      # await asyncio.sleep(2)
      time.sleep(5)
      print("corou 2, {}".format(time.perf_counter()))
  
  async def corou3():
      print("corou 3, {}".format(time.perf_counter()))
      await asyncio.sleep(2)
      print("corou 3, {}".format(time.perf_counter()))
  
  async def main():
      tasks = [asyncio.create_task(corou())for corou in [corou1,corou2,corou3] ]
      await asyncio.gather(*tasks)
  
  if __name__ == "__main__":
      asyncio.run(main())
  ###########执行结果############
  func 1, 0.3864962
  func 2, 0.3866095
  func 2, 5.386688
  func 3, 5.387335
  func 1, 5.3877248
  func 3, 7.3883411
  
  通过上述例子可以看出，协程corou1，corou2，corou3的执行顺序按照加入循环事件的顺序。协程corou1 通过 await asyncio.sleep(2) 让出控制权，协程corou2由于没有await让出控制权所以一直在执行。当corou2结束后，很显然corou1也已经执行完，但还是先执行协程corou3，等协程corou3通过await让出控制权后，协程corou1才执行。
  
  从上可以看出：
  - 协程是按照事件循环顺序执行。即使交出控制权的协程执行完毕任务后等待再次获取控制权完成剩余任务，也必须等到循环到该协程才能获得CPU控制权。不会由于该协程已经完成await的任务而直接给它。这就是解释，协程corou2执行完会继续执行协程corou3，而并非继续执行协程corou1。
  - await是交出协程控制权的地方
  - 从上述代码可以看出，协程中当执行耗时IO操作时，应及时释放控制权，否则其他协程即便完成任务也由于没有CPU控制权而无法继续执行任务。

  2019-07-09

  

   1
• 

  Marj

  在 jupyter 中运行，报
  ```
  ---------------------------------------------------------------------------
  AttributeError Traceback (most recent call last)
  <timed eval> in <module>
  
  AttributeError: module 'asyncio' has no attribute 'run'
  ```

  2019-07-02

  

   1