Python进阶:Asyncio并发编程

一、Sync & Async

• Sync(同步):是指操作一个接一个地执行,下一个操作必须等上一个操作完成后才能执行
• Async(异步):是指不同操作间可以相互交替执行,如果其中的某个操作被block了,程序并不会等待,而是会找出可执行的操作继续执行

二、什么是Asyncio?

• 和其它Python程序一样,是单线程的,只有一个主线程,但是可以进行多个不同的任务(task),这里的任务,就是特殊的future对象,这些不同的任务,被一个叫做event loop的对象所控制。可以把这里的任务,类比多线程版本里的多个线程

先来看一段简单的代码:

import time
def crawl_page(url):
	print('crawling {}'.format(url))
	sleep_time = int(url.split('_')[-1])
	time.sleep(sleep_time)
	print('OK {}'.format(url))

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

if __name__ == '__main__':
    start_time = time.perf_counter()
    main(['url_1','url_2','url_3','url_4'])
    end_time = time.perf_counter()
    print('run {} seconds'.format(end_time - start_time))

上述代码是爬虫的简单示例,四个url一共用了10s的时间,如何用协程进行优化呢?

import time
import asyncio

async def crawl_page(url):
	print('crawling {}'.format(url))
	sleep_time = int(url.split('_')[-1])
	time.sleep(sleep_time)
	print('OK {}'.format(url))

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

if __name__ == '__main__':
    start_time = time.perf_counter()
    asyncio.run(main(['url_1','url_2','url_3','url_4']))
    end_time = time.perf_counter()
    print('run {} seconds'.format(end_time - start_time))

• async:声明异步函数,上面的crawl_page和main都成了异步函数
• await:协程的关键字,后面接异步函数

1、当遇到 await 关键字时，协程会暂停自己的执行。
2、执行 await 后面的异步函数调用，将控制权交给异步函数。
3、异步函数开始执行异步操作，可能会暂时挂起自己的执行，等待异步操作完成。
4、在异步操作完成后，异步函数会恢复执行，并返回结果。
5、协程接收到异步函数的返回结果，恢复执行 await 语句后面的代码。
总结一下就是两个作用:
a、暂停当前协程的执行，将控制权交给事件循环（Event Loop），允许其他任务继续执行。在等待的过程中，协程不会阻塞整个程序或线程，而是允许其他协程或任务继续执行
b、等待异步操作的完成，并获取其返回的结果

• asyncio.run:作为主程序的入口函数,在程序运行周期内,只调用一次asyncio.run

运行上述代码发现还需要10s,原因是await关键字是同步调用,crawl_page(url)在当前的调用结束之前,是不会触发下一次调用的,相当于用异步接口写了同步代码

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

if __name__ == '__main__':
    start_time = time.perf_counter()
    asyncio.run(main(['url_1','url_2','url_3','url_4']))
    end_time = time.perf_counter()
    print('run {} seconds'.format(end_time - start_time))

上述代码只需要4s左右,运行总时长等于运行时间最长的爬虫

• asyncio.create_task():返回一个Task对象,Task对象表示一个可调度的协程任务,可以被事件循环调度和执行,有了task对象之后,协程任务才算真正生效

对于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]
    await asyncio.gather(*tasks)

if __name__ == '__main__':
    start_time = time.perf_counter()
    asyncio.run(main(['url_1','url_2','url_3','url_4']))
    end_time = time.perf_counter()
    print('run {} seconds'.format(end_time - start_time))

• asyncio.gather:并发执行多个协程任务的函数,接受一系列的协程对象,并返回一个协程对象,该协程对象表示所有任务的集合
• *tasks:解包列表,将列表转换为函数的参数;与之对应的 * *dict可以将字典转化为函数的参数

其它用法:

import asyncio


async def worker_1():
        await asyncio.sleep(1)
        return 1

async def worker_2():
        await  asyncio.sleep(2)
        return 2 / 0

async def worker_3():
        await asyncio.sleep(3)
        return 3

async def main():
        task_1 = asyncio.create_task(worker_1())
        task_2 = asyncio.create_task(worker_2())
        task_3 = asyncio.create_task(worker_3())

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

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

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

• cancal():取消方法,可取消任务
• return_exceptions = True:如果不设置这个参数,错误就会完整地throw到执行层,从而需要try…catch来捕获,这也就意味着其它还没被执行的任务会被全部取消掉
• res:上面说过await可以获取异步操作返回的结果

三、Asyncio工作原理

假设任务只有两个状态:一是预备状态;二是等待状态。所谓预备状态,是指任务目前空闲,但随时待命准备运行。而等待状态,是指任务已经运行,而正在等待外部的操作完成,比如I/O操作
在这种情况下,event loop会维护两个任务列表,分别对应这两种状态;并且选取预备状态的一个任务,使其运行,一直到把这个任务把控制权交还给event loop为止

当任务把控制权交还给event loop时,event loop会根据其是否完成,把任务放到预备或等待状态的列表,然后遍历等待状态列表的任务,查看它们是否完成，如果完成,则将其放到预备状态的列表,如果未完成,则继续放在等待状态的列表，如此周而复始,直到所有的任务都完成

四、Asyncio与多线程区别

• ASyncio可以在不使用多线程的情况下实现并发编程,而且程序运行过程中不容易被打断,不容易出现race condition
• 但是实际情况下,想用好Asyncio,很多情况下必须得有相应的Python库支持,Asyncio软件库的兼容性是个问题
• 对于和多线程的区别,下面是个总结

如果是I/O heavy,并且I/O操作很慢,需要很多任务/线程协同运行,那么使用Asyncio更合适;
如果是I/O heavy,但是I/O操作很快,只需要有限数量的任务/线程,那么使用多线程就可以
如果是CPU heavy,则可以使用多进程来解决问题