Python核心技术与实战——十五|Python协程

Posted 银色的音色

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了Python核心技术与实战——十五|Python协程相关的知识，希望对你有一定的参考价值。

我们在上一章将生成器的时候最后写了，在Python2中生成器还扮演了一个重要的角色——实现Python的协程。那什么是协程呢？

协程

协程是实现并发编程的一种方式。提到并发，肯很多人都会想到多线程/多进程模型，这就是解决并发问题的经典模型之一。在最初的互联网世界中，多线程/多进程就在服务器并发中起到举足轻重的作用。

但是随着互联网的发展，慢慢很多场合都会遇到C10K瓶颈，也就是同时连接到服务器的客户达到1W，于是，很多代码就跑崩溃，因为进程的上下文切换占用了大量的资源，线程也顶不住如此巨大的压力。这时候，NGINX就带着事件循环闪亮登场了。

事件循环启动一个统一的调度器，让调度器来决定一个时刻去运行哪个任务，于是省却了多线程中启动线程、管理线程、同步锁等各种开销。同一时期的NGINX，在高并发下也能保持资源消耗低、性能高，相比Apache也支持更多的并发连接。

再到后来，出现了一个很有名的名词——回调地狱(callback hell).很多写过JavaScript的朋友明白那是什么。大家惊喜的发现，这种工具很完美的继承了事件循环的优越性，同时还能提供async/await语法糖，解决了执行性和可读性共存的难题。于是，协程渐渐被更多人发现并看好，也有越来越多的人尝试用Node.js做起了后端开发。

回到Python中，使用生成器来实现协程已经是Python2时代的老方法了。而在Python3.7中，提供了基于asynico和async/await的方法。我们这节课就抛弃生成器的方法，基于这种用法来实现协程。

从一个爬虫说起

爬虫的作用就不多讲 了，我们直接看代码

import time
def craw_page(url):
    print(‘crawing{}‘.format(url))
    sleep_time = int(url.split(‘_‘)[-1])
    time.sleep(sleep_time)
    print(‘OK {}‘.format(url))

def main(urls):
    for url in urls:
        craw_page(url)

main([‘url_1‘,‘url_2‘,‘url_3‘,‘url_4‘])

我们通过上面的代码依次爬取了5个页面，而每个页面爬取的时间分别为1-4秒（我们用time.sleep模拟了数据抓取的过程）。所以整个程序总耗时为

1+2+3+4= 10秒

程序的时间基本上都消耗在等待上了。那我们是不是可以怎么优化一下呢？一个很简单的思路出现了：我们这种操作可以并发化，就让我们看看怎么写

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))

async def main(urls):
    for url in urls:
        await crawl_page(url)


time_start = time.perf_counter()
asyncio.run(main([‘url_1‘,‘url_2‘,‘url_3‘,‘url_4‘]))
print(‘totle cost {}s‘.format(time.perf_counter()-time_start))

在这段代码里就实现了使用协程写异步程序的简便方法。

import asyncio

这个库里就包含了我们需要的大部分实现协程的工具。

而async修饰词声明异步函数，于是这里的两个函数都成了异步函数。而调用的异步函数我们就得到了一个协程对象(coroutine object)

然后通过awaitlai来调用。而await的执行结果和Python的正常执行是一样的。也会使程序阻塞在此处，进入被调用的函数，直到执行完毕后再返回继续。这也就是await字面上的意思。

这里就要我们来看看执行协程的几种方法了,一般常用的方法有三种：

1.通过await来调用，就和上面说的一样代码中await asyncio.sleep(sleep_time)处会模拟爬虫获取数据时等待的时间，而await crawl_page()则会执行craw_page()函数。

2.我们可以通过

asyncio.create_task()

方法来创建任务，具体的方法我们在下一章会说，这里先点明一下

3.我们需要让asyncio.run来触发运行，这个函数实在Python3.7以后的版本中才有的特性，可以让Python协程接口编程变得非常简单，我们不用理会事件循环怎么定义和如何使用的问题(我们在下面会讲到)。而且一个非常好的编程规范是

asyncio.run(main())

作为主程序的入口，在程序的运行周期内，知掉用一次该函数。

这样，我们就可以跑一下上面的代码，看看结论是什么！

############运行结论############
crawlingurl_1
OK url_1
crawlingurl_2
OK url_2
crawlingurl_3
OK url_3
crawlingurl_4
OK url_4
totle cost 10.0089496s

为什么还是10s呢？没错，上面讲过了，await是同步调用（上面字体加粗的部分）。因此，craw_page()函数在当前的调用结束前是不会触发下一次调用的。于是代码效果就和上面完全一样了。相当于用异步接口写了个同步代码

那又该怎么办呢？

其实很简单，也是我们在下面要讲的东西——任务(Task)。老规矩，我们通过下面的代码来讲解

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))

async def main(urls):
    tasks = [(asyncio.create_task(crawl_page(url))) for url in urls]
    for task in tasks:
        await task


time_start = time.perf_counter()
asyncio.run(main([‘url_1‘,‘url_2‘,‘url_3‘,‘url_4‘]))
print(‘totle cost {}s‘.format(time.perf_counter()-time_start))

可以看到，我们有了协程对象以后，可以通过

asyncio.creat_task()

来创建任务，任务创建后会被调度执行，这样，我们的代码也不会被阻塞在任务这里，所以我们要等到所有任务都结束才行，用循环启动tasks里的task

这样一来，运行结果就不一样了

##########运行结论##########
crawlingurl_1
crawlingurl_2
crawlingurl_3
crawlingurl_4
OK url_1
OK url_2
OK url_3
OK url_4
totle cost 4.0060087s

程序运行总时长等于运行时间最长的爬虫的运行时间。

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

 

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))

async def main(urls):
    tasks = [(asyncio.create_task(crawl_page(url))) for url in urls]
    for task in tasks:
        await asyncio.gather(*tasks)


time_start = time.perf_counter()
asyncio.run(main([‘url_1‘,‘url_2‘,‘url_3‘,‘url_4‘]))
print(‘totle cost {}s‘.format(time.perf_counter()-time_start))

我们通过*tasks来解包列表，将列表变成了函数的参数，与之对应的是**dict是将字典变成函数的参数（和）

上面已经大致说明的协程的用法，如果需要爬取的页有上万个又该怎么办呢？再对比一下协程的写法，是不是这种写法更加清晰明了。

 

解密协程运行时

下面我们可以深入底层代码看看协程工作期间是如何运行的，还是先放两段代码

import asyncio
import time
async def work_1():
    print(‘work_1 start‘)
    await asyncio.sleep(1)
    print(‘work_1 done‘)

async def work_2():
    print(‘work_1 start‘)
    await asyncio.sleep(2)
    print(‘work_2 done‘)

async def main():
    print(‘befort await‘)
    await work_1()
    print(‘awaited work_1‘)
    await work_2()
    print(‘awaited work_2‘)
    

start_time = time.perf_counter()
asyncio.run(main())
print(‘totle cost:{}s‘.format(time.perf_counter()-start_time))

##########输出##########
befort await
work_1 start
work_1 done
awaited work_1
work_1 start
work_2 done
awaited work_2
totle cost:3.0037941s

代码段2

import asyncio
import time

async def work_1():
    print(‘work_1 start‘)
    await asyncio.sleep(1)
    print(‘work_1 done‘)

async def work_2():
    print(‘work_1 start‘)
    await asyncio.sleep(2)
    print(‘work_2 done‘)

async def main():
    task1 = asyncio.create_task(work_1())
    task2 = asyncio.create_task(work_2())
    print(‘before await‘)
    await task1
    print(‘awaited work_1‘)
    await task2
    print(‘awaited work_2‘)
    
start_time = time.perf_counter()
asyncio.run(main())
print(‘totle cost:{}s‘.format(time.perf_counter()-start_time))

##########输出##########
before await
work_1 start
work_1 start
work_1 done
awaited work_1
work_2 done
awaited work_2
totle cost:2.0024394s

我们分析一下整个流程来更加详细的了解协程和线程的具体区别：

1.asyncio.run(main())进入main()函数，事件循环开启；

2.task1和task2两个任务被创建，并且进入事件循环等待运行；程序执行到第一个print，输入‘before await‘字符串

3.task1被await执行，用户选择从当前的主任务中切出，事件调度器开始调度work_1;

4.work_1开始运行，先执行print输出work_1 start后运行到等待后从当前任务切出，事件调度器开始调度work_2；

5.work_2开始运行，执行print输出work_2 start后到await处从当前任务切出

6.上述所有事件的运行时间都应该在1-10ms之间，甚至会更短，事件调度器从这个时刻开始暂停调度

7.1s后，work_1的sleep完成，时间调度器将控制权重新传给task1，输出work_1 done 后work_1完成任务，，从事件循环中退出；

8.await task1事件完成，事件调度器将控制器传给主任务，输出awaited work_1。然后在await task2处继续等待；

9.两秒钟后，work_2的sleep完成，事件调度器把控制器传给task2，输出work_2 done。task2完成任务从事件循环中退出；

10。主任务输出await work_2，协程任务结束，事件循环结束。

 

上面讲了协程的基本用法，可是还有些应用场景是需要有一些附加条件的：比方给协程任务限定一个运行时间，如果超出时间就取消，或者某些协程运行时出现了错误，那该怎么处理呢？我们看看下面的代码

import asyncio
import time
async def work_1():
    await asyncio.sleep(1)
    return 1

async def work_2():
    await asyncio.sleep(2)
    return 2/0       #除数不能为0，这里制造出一个错误

async def work_3():
    await asyncio.sleep(10)
    return 3

async def main():
    task_1 = asyncio.create_task(work_1())
    task_2 = asyncio.create_task(work_2())
    task_3 = asyncio.create_task(work_3())

    await asyncio.sleep(3)
    task_3.cancel()

    res = await asyncio.gather(task_1,task_2,task_3,return_exceptions = True)
    print(res)

start_time = time.perf_counter()
asyncio.run(main())
print(‘totle cost:{}s‘.format(time.perf_counter()-start_time))

##########输出##########
[1, ZeroDivisionError(‘division by zero‘), CancelledError()]
totle cost:3.00382s

我们可以看到，work_1可以正常运行，work_2在运行过程出现错误，work_3执行时间过长被cancel掉，这些信心都会体现在最后的结果res中，

并且还有一点疑问，我们只是在return中设置 了一处错误，可是如果程序过程出现错误又是什么效果呢？我们把work_2修改一下看看效果

async def work_2():
    l = [1,2,3]
    l[4]
    await asyncio.sleep(2)
    return 2       


##########输出##########
[1, IndexError(‘list index out of range‘), CancelledError()]
totle cost:3.0029349s

并不是只有ruturn里有错误才会把错误返回，而是只要程序有错误都会把错误返回给主任务。但是我们必须加上return_exception = True这个条件，否则错误就会被完整的throw到执行层从而需要用try except的方式来捕捉，那么也就意味着其他还没有执行的任务会被全部取消掉。 

讲到这里，我们就可以发现，线程能偶实现的，协程也都可做到。那么我们就温习一下上面的知识点，用协程来做一个生产者消费者模型吧

import asyncio
import random
async def consumer(queue,id):
    while True:
        val = await queue.get()
        print(‘{} get a val:{}.‘.format(id,val))
        await asyncio.sleep(1)

async def producer(queue,id):
    for i in range(5):
        val = random.randint(4,20)
        await queue.put(val)
        print(‘{} put a val:{}.‘.format(id,val))
        await asyncio.sleep(1)

async def main():
    queue = asyncio.Queue()

    consumer_1 = asyncio.create_task(consumer(queue,‘consumer_1‘))
    # consumer_2 = asyncio.create_task(consumer(queue,‘consumer_2‘))

    producer_1 = asyncio.create_task(producer(queue,‘producer_1‘))
    # producer_2 = asyncio.create_task(producer(queue,‘producer_2‘))


    await asyncio.sleep(10)
    consumer_1.cancel()
    producer_1.cancel()

    await asyncio.gather(consumer_1,producer_1,return_exceptions=True)

asyncio.run(main())

我们定义了一个生产者和一个消费者，在main里启动了一个生产者一个消费者。并且要求10s后cancel掉消费者（其实生产者在for里已经定义了只通过5次循环生产出来5个元素）。

并且在主任务里的通过sleep规定了主任务的运行时长，不管是否还有任务在执行都通过后面的代码cancel掉。

实战

最后我们通过一个完整的爬虫来进行今天的实战练习

我们通过一个页面：https://movie.douban.com/cinema/later/xian/，这个页面描述了西安最近上映的电影，那如何通过python获取到这些电影的名称’、上映时间和海报呢？

这点我们留着后面完善。

总结

到这里，今天的内容就讲完了，今天用了较长的篇幅从一个简单的爬虫到一个真正的爬虫之间穿插讲述了Python协程比较新的方法和概念。这里复习一下：

协程和多线程的区别在于两点：1.协程为单线程，2.协程有用户决定在哪里交出控制权切换到下一个任务

协程的写法更加简洁清晰，把async/await的语法和create_task结合来用，对于中小级别的开发需求已经毫无压力

写协程程序的时候，大脑里应该有个清晰的事件循环概念，知道在什么时候需要暂停、等待IO、什么时候需要一并执行到底。

最后要记得，什么时候用什么模型能能 达到工程上的最优，而不是觉得那个技术非常牛，就创造条件上该技术。总之一句话：

技术是工程，而工程则是时间、资源、人力等众多纷繁复杂的事情的折中。

最后想一想：

协程是如何实现回调函数的呢？

以上是关于Python核心技术与实战——十五|Python协程的主要内容，如果未能解决你的问题，请参考以下文章

Python核心技术与实战

Python 核心技术与实战 --01 列表与元祖

Python核心技术与实战——十三|Python中参数传递机制

Python 核心技术与实战 --02 字典和集合

Python自然语言处理实战核心技术与算法，Python自然语言处理，PyTorch深度学习实战下载

《Python自然语言处理实战：核心技术与算法》PDF新书推介，附带链接地址