【Python核心】揭秘Python协程

首先要明白什么是协程?

协程是实现并发编程的一种方式。一说并发肯定想到了多线程/多进程模型,多线程/多进程正是解决并发问题的经典模型之一

先从一个爬虫实例出发,用清晰的思路并且结合实战来搞懂这个不算特别容易理解的概念。之后,再由浅入深直击协程的核心

一、从一个爬虫说起

爬虫,就是互联网的蜘蛛,在搜索引擎诞生时与其一同来到世上
爬虫每秒钟都会爬取大量的网页,提取关键信息后存储在数据库中以便日后分析。爬虫有非常简单的Python十行代码实现,也有Google那样的全球分布式爬虫的上百万行代码,分布在内部上万台服务器上对全世界的信息进行嗅探

先看一个简单的爬虫例子:

1import time 2 3def crawl_page(url): 4 print('crawling {}'.format(url)) 5 sleep_time = int(url.split('_')[-1]) 6 time.sleep(sleep_time) 7 print('OK {}'.format(url)) 8 9def main(urls): 10 for url in urls: 11 crawl_page(url) 12 13%time main(['url_1', 'url_2', 'url_3', 'url_4']) 14 15########## 输出 ########## 16 17crawling url_1 18OK url_1 19crawling url_2 20OK url_2 21crawling url_3 22OK url_3 23crawling url_4 24OK url_4 25Wall time: 10 s 26 27

注意:
本节的主要目的是协程的基础概念,因此简化爬虫的scrawl_page函数为休眠数秒,休眠时间取决于url最后的那个数字

这是一个很简单的爬虫,main()函数执行时调取crawl_page()函数进行网络通信,经过若干秒等待后收到结果,然后执行下一个

看起来很简单,但仔细一算它也占用了不少时间,五个页面分别用了1秒到4秒的时间,加起来一共用了10秒。这显然效率低下,该怎么优化呢?

二、简单的协程示例

于是,一个很简单的思路出现了——这种爬取操作完全可以并发化。接下来看看使用协程怎么写

1import asyncio 2 3async def crawl_page(url): 4 print('crawling {}'.format(url)) 5 sleep_time = int(url.split('_')[-1]) 6 await asyncio.sleep(sleep_time) 7 print('OK {}'.format(url)) 8 9async def main(urls): 10 for url in urls: 11 await crawl_page(url) 12 13%time asyncio.run(main(['url_1', 'url_2', 'url_3', 'url_4'])) 14 15########## 输出 ########## 16 17crawling url_1 18OK url_1 19crawling url_2 20OK url_2 21crawling url_3 22OK url_3 23crawling url_4 24OK url_4 25Wall time: 10 s 26 27

看到这段代码,可以发现在Python 3.7以上版本中,使用协程写异步程序非常简单

首先来看import asyncio,这个库包含了大部分我们实现协程所需的魔法工具
sync修饰词声明异步函数,于是,这里的crawl_page和main都变成了异步函数。而调用异步函数,便可得到一个协程对象(coroutine object)

举个例子,如果print(crawl_page(''))便会输出提示这是一个 Python的协程对象,而并不会真正执行这个函数

三、协程的执行

再来说说协程的执行
执行协程有多种方法,这里介绍一下常用的三种

首先,可以通过await来调用
await执行的效果和Python正常执行是一样的,也就是说程序会阻塞在这里进入被调用的协程函数,执行完毕返回后再继续,而这也是await的字面意思。代码中await asyncio.sleep(sleep_time)会在这里休息若干秒,await crawl_page(url)则会执行crawl_page()函数

其次,可以通过asyncio.create_task()来创建任务

最后,需要asyncio.run来触发运行
asyncio.run这个函数是Python 3.7之后才有的特性,可以让Python的协程接口变得非常简单,不用去理会事件循环怎么定义和怎么使用的问题
一个非常好的编程规范是,asyncio.run(main())作为主程序的入口函数,在程序运行周期内只调用一次asyncio.run

这样,大概看懂了协程是怎么用的吧。不妨试着跑一下代码,怎么还是10秒?

10 秒就对了,如上面所说await是同步调用,因此,crawl_page(url)在当前的调用结束之前是不会触发下一次调用的。于是,这个代码效果就和上面完全一样,相当于用异步接口写了个同步代码

现在又该怎么办呢?

其实很简单,也正是接下来要讲的协程中的一个重要概念,任务(Task)。继续看下面的代码

1import asyncio 2 3async def crawl_page(url): 4 print('crawling {}'.format(url)) 5 sleep_time = int(url.split('_')[-1]) 6 await asyncio.sleep(sleep_time) 7 print('OK {}'.format(url)) 8 9async def main(urls): 10 tasks = [asyncio.create_task(crawl_page(url)) for url in urls] 11 for task in tasks: 12 await task 13 14%time asyncio.run(main(['url_1', 'url_2', 'url_3', 'url_4'])) 15 16########## 输出 ########## 17 18crawling url_1 19crawling url_2 20crawling url_3 21crawling url_4 22OK url_1 23OK url_2 24OK url_3 25OK url_4 26Wall time: 3.99 s 27 28

可以看到,有了协程对象后便可以通过asyncio.create_task来创建任务,任务创建后很快就会被调度执行,这样,代码也不会阻塞在任务这里。所以,要等所有任务都结束才行,用for task in tasks: await task即可

可以看到执行的效果,结果显示运行总时长等于运行时间最长的爬虫

当然,也可以想一想这里用多线程应该怎么写?而如果需要爬取的页面有上万个又该怎么办呢?再对比下协程的写法,谁更清晰自是一目了然

其实,对于执行tasks还有另一种做法:

1import asyncio 2 3async def crawl_page(url): 4 print('crawling {}'.format(url)) 5 sleep_time = int(url.split('_')[-1]) 6 await asyncio.sleep(sleep_time) 7 print('OK {}'.format(url)) 8 9async def main(urls): 10 tasks = [asyncio.create_task(crawl_page(url)) for url in urls] 11 await asyncio.gather(*tasks) 12 13%time asyncio.run(main(['url_1', 'url_2', 'url_3', 'url_4'])) 14 15########## 输出 ########## 16 17crawling url_1 18crawling url_2 19crawling url_3 20crawling url_4 21OK url_1 22OK url_2 23OK url_3 24OK url_4 25Wall time: 4.01 s 26 27

这里的代码也很好理解,唯一要注意的是*tasks解包列表,将列表变成了函数的参数,与之对应的是**dict将字典变成了函数的参数

注意:
asyncio.create_task,asyncio.run这些函数都是Python 3.7以上的版本才提供的

四、解密协程运行时

不妨来深入代码底层看看。有了前面的知识做基础应该很容易理解这两段代码

1import asyncio 2 3async def worker_1(): 4 print('worker_1 start') 5 await asyncio.sleep(1) 6 print('worker_1 done') 7 8async def worker_2(): 9 print('worker_2 start') 10 await asyncio.sleep(2) 11 print('worker_2 done') 12 13async def main(): 14 print('before await') 15 await worker_1() 16 print('awaited worker_1') 17 await worker_2() 18 print('awaited worker_2') 19 20%time asyncio.run(main()) 21 22########## 输出 ########## 23 24before await 25worker_1 start 26worker_1 done 27awaited worker_1 28worker_2 start 29worker_2 done 30awaited worker_2 31Wall time: 3 s 32 33
1import asyncio 2 3async def worker_1(): 4 print('worker_1 start') 5 await asyncio.sleep(1) 6 print('worker_1 done') 7 8async def worker_2(): 9 print('worker_2 start') 10 await asyncio.sleep(2) 11 print('worker_2 done') 12 13async def main(): 14 task1 = asyncio.create_task(worker_1()) 15 task2 = asyncio.create_task(worker_2()) 16 print('before await') 17 await task1 18 print('awaited worker_1') 19 await task2 20 print('awaited worker_2') 21 22%time asyncio.run(main()) 23 24########## 输出 ########## 25 26before await 27worker_1 start 28worker_2 start 29worker_1 done 30awaited worker_1 31worker_2 done 32awaited worker_2 33Wall time: 2.01 s 34 35

不过,第二个代码到底发生了什么呢?为了更详细了解到协程和线程的具体区别,这里详细地分析了整个过程

  1. asyncio.run(main()),程序进入main()函数,事件循环开启
  2. task1和task2任务被创建,并进入事件循环等待运行,运行到print时输出before await
  3. await task1执行,用户选择从当前的主任务中切出,事件调度器开始调度worker_1
  4. worker_1开始运行,运行print输出worker_1 start,然后运行到await asyncio.sleep(1), 从当前任务切出,事件调度器开始调度worker_2
  5. worker_2开始运行,运行print输出worker_2 start,然后运行到await asyncio.sleep(2),从当前任务切出
  6. 以上所有事件的运行时间,都应该在1ms到10ms之间,甚至可能更短,事件调度器从这个时候开始暂停调度
  7. 一秒钟后,worker_1的sleep完成,事件调度器将控制权重新传给task_1,输出worker_1 done,task_1完成任务,从事件循环中退出
  8. await task1完成,事件调度器将控制器传给主任务,输出awaited worker_1,·然后在await task2处继续等待
  9. 两秒钟后,worker_2的sleep完成,事件调度器将控制权重新传给task_2,输出worker_2 done,task_2完成任务,从事件循环中退出
  10. 主任务输出awaited worker_2,协程全任务结束,事件循环结束

接下来,进阶一下。如果想给某些协程任务限定运行时间,一旦超时就取消,又该怎么做呢?再进一步,如果某些协程运行时出现错误,又该怎么处理呢?同样的,来看代码

1 2import asyncio 3 4async def worker_1(): 5 await asyncio.sleep(1) 6 return 1 7 8async def worker_2(): 9 await asyncio.sleep(2) 10 return 2 / 0 11 12async def worker_3(): 13 await asyncio.sleep(3) 14 return 3 15 16async def main(): 17 task_1 = asyncio.create_task(worker_1()) 18 task_2 = asyncio.create_task(worker_2()) 19 task_3 = asyncio.create_task(worker_3()) 20 21 await asyncio.sleep(2) 22 task_3.cancel() 23 24 res = await asyncio.gather(task_1, task_2, task_3, return_exceptions=True) 25 print(res) 26 27%time asyncio.run(main()) 28 29########## 输出 ########## 30 31[1, ZeroDivisionError('division by zero'), CancelledError()] 32Wall time: 2 s 33 34

可以看到,worker_1正常运行,worker_2运行中出现错误,worker_3执行时间过长被cancel,这些信息全部体现在最终的返回结果res中

不过要注意return_exceptions=True这行代码。如果不设置这个参数错误就会完整地throw到执行层,从而需要try except来捕捉,这也就意味着其他还没被执行的任务会被全部取消掉。为了避免这个局面,将return_exceptions设置为True即可

到这里,可以发现线程能实现的,协程都能做到。接下来用协程来实现一个经典的生产者消费者模型

1import asyncio 2import random 3 4async def consumer(queue, id): 5 while True: 6 val = await queue.get() 7 print('{} get a val: {}'.format(id, val)) 8 await asyncio.sleep(1) 9 10async def producer(queue, id): 11 for i in range(5): 12 val = random.randint(1, 10) 13 await queue.put(val) 14 print('{} put a val: {}'.format(id, val)) 15 await asyncio.sleep(1) 16 17async def main(): 18 queue = asyncio.Queue() 19 20 consumer_1 = asyncio.create_task(consumer(queue, 'consumer_1')) 21 consumer_2 = asyncio.create_task(consumer(queue, 'consumer_2')) 22 23 producer_1 = asyncio.create_task(producer(queue, 'producer_1')) 24 producer_2 = asyncio.create_task(producer(queue, 'producer_2')) 25 26 await asyncio.sleep(10) 27 consumer_1.cancel() 28 consumer_2.cancel() 29 30 await asyncio.gather(consumer_1, consumer_2, producer_1, producer_2, return_exceptions=True) 31 32%time asyncio.run(main()) 33 34########## 输出 ########## 35 36producer_1 put a val: 5 37producer_2 put a val: 3 38consumer_1 get a val: 5 39consumer_2 get a val: 3 40producer_1 put a val: 1 41producer_2 put a val: 3 42consumer_2 get a val: 1 43consumer_1 get a val: 3 44producer_1 put a val: 6 45producer_2 put a val: 10 46consumer_1 get a val: 6 47consumer_2 get a val: 10 48producer_1 put a val: 4 49producer_2 put a val: 5 50consumer_2 get a val: 4 51consumer_1 get a val: 5 52producer_1 put a val: 2 53producer_2 put a val: 8 54consumer_1 get a val: 2 55consumer_2 get a val: 8 56Wall time: 10 s 57 58

五、实战豆瓣近日推荐电影爬虫

接下来进入实战环节,实现一个完整的协程爬虫

任务描述:https://movie.douban.com/cinema/later/beijing/这个页面描述了北京最近上映的电影,通过Python得到这些电影的名称、上映时间和海报?这个页面的海报是缩小版的,希望从具体的电影描述页面中抓取到海报

下面给出了同步版本的代码和协程版本的代码,通过运行时间和代码写法的对比,希望能对协程有更深的了解(注意:为了突出重点和简化代码,因此省略了异常处理)

1import requests 2from bs4 import BeautifulSoup 3 4def main(): 5 url = "https://movie.douban.com/cinema/later/beijing/" 6 init_page = requests.get(url).content 7 init_soup = BeautifulSoup(init_page, 'lxml') 8 9 all_movies = init_soup.find('div', id="showing-soon") 10 for each_movie in all_movies.find_all('div', class_="item"): 11 all_a_tag = each_movie.find_all('a') 12 all_li_tag = each_movie.find_all('li') 13 14 movie_name = all_a_tag[1].text 15 url_to_fetch = all_a_tag[1]['href'] 16 movie_date = all_li_tag[0].text 17 18 response_item = requests.get(url_to_fetch).content 19 soup_item = BeautifulSoup(response_item, 'lxml') 20 img_tag = soup_item.find('img') 21 22 print('{} {} {}'.format(movie_name, movie_date, img_tag['src'])) 23 24%time main() 25 26########## 输出 ########## 27 28阿拉丁 0524日 https://img3.doubanio.com/view/photo/s_ratio_poster/public/p2553992741.jpg 29龙珠超:布罗利 0524日 https://img3.doubanio.com/view/photo/s_ratio_poster/public/p2557371503.jpg 30五月天人生无限公司 0524日 https://img3.doubanio.com/view/photo/s_ratio_poster/public/p2554324453.jpg 31... ... 32直播攻略 0604日 https://img3.doubanio.com/view/photo/s_ratio_poster/public/p2555957974.jpg 33Wall time: 56.6 s 34 35
1import asyncio 2import aiohttp 3 4from bs4 import BeautifulSoup 5 6async def fetch_content(url): 7 async with aiohttp.ClientSession( 8 headers=header, connector=aiohttp.TCPConnector(ssl=False) 9 ) as session: 10 async with session.get(url) as response: 11 return await response.text() 12 13async def main(): 14 url = "https://movie.douban.com/cinema/later/beijing/" 15 init_page = await fetch_content(url) 16 init_soup = BeautifulSoup(init_page, 'lxml') 17 18 movie_names, urls_to_fetch, movie_dates = [], [], [] 19 20 all_movies = init_soup.find('div', id="showing-soon") 21 for each_movie in all_movies.find_all('div', class_="item"): 22 all_a_tag = each_movie.find_all('a') 23 all_li_tag = each_movie.find_all('li') 24 25 movie_names.append(all_a_tag[1].text) 26 urls_to_fetch.append(all_a_tag[1]['href']) 27 movie_dates.append(all_li_tag[0].text) 28 29 tasks = [fetch_content(url) for url in urls_to_fetch] 30 pages = await asyncio.gather(*tasks) 31 32 for movie_name, movie_date, page in zip(movie_names, movie_dates, pages): 33 soup_item = BeautifulSoup(page, 'lxml') 34 img_tag = soup_item.find('img') 35 36 print('{} {} {}'.format(movie_name, movie_date, img_tag['src'])) 37 38%time asyncio.run(main()) 39 40########## 输出 ########## 41 42阿拉丁 0524日 https://img3.doubanio.com/view/photo/s_ratio_poster/public/p2553992741.jpg 43龙珠超:布罗利 0524日 https://img3.doubanio.com/view/photo/s_ratio_poster/public/p2557371503.jpg 44五月天人生无限公司 0524日 https://img3.doubanio.com/view/photo/s_ratio_poster/public/p2554324453.jpg 45... ... 46直播攻略 0604日 https://img3.doubanio.com/view/photo/s_ratio_poster/public/p2555957974.jpg 47Wall time: 4.98 s 48 49

六、总结

用了较长的篇幅,从一个简单的爬虫开始,到一个真正的爬虫结束,在中间穿插讲解了Python协程最新的基本概念和用法

  • 协程和多线程的区别

主要在于两点,一是协程为单线程,二是协程由用户决定,在哪些地方交出控制权,切换到下一个任务

  • 协程的写法更加简洁清晰

把async/await语法和create_task结合来用,对于中小级别的并发需求毫无压力

  • 清晰的事件循环概念

写协程程序的时候,要有清晰的事件循环概念,知道程序在什么时候需要暂停、等待 I/O,什么时候需要一并执行到底

最后的最后,请一定不要轻易炫技
多线程模型也一定有其优点,一个真正牛逼的程序员应该懂得,在什么时候用什么模型能达到工程上的最优

代码交流 2021