Python并发编程系列之协程

python基础

浏览数:318

2019-3-25

1 引言

  协程是近几年并发编程的一个热门话题,与Python多进程、多线程相比,协程在很多方面优势明显。本文从协程的定义和意义出发,结合asyncio模块详细讲述协程的使用。

2 协程的意义

2.1 什么是协程

  协程,又称微线程,英文名为Coroutine。对于多线程,在执行一个个不同任务时,遇到阻塞(例如IO操作)时,操作系统会自动将CPU资源切换给另一个线程。但协程不同,协程是用户态的轻量级线程,更多的依靠用户切换。

2.2 协程的作用

         与线程不同的是,协程需要用户自己进行手动切换——当某线程在执行任务中的函数A(协程A))时,可任意终端,手动切换到任务中的另一个函数B(协程B),然后在适当的时候在回到函数A(协程A)中继续执行,这样虽然繁琐,但也提供了更大的操作自由度,同时协程A和协程B都属于同一线程,切换效率相比于线程或进程间的切换有极大地优势。另外,协程不需要多线程的锁机制,因为都属于同一个线程,也不存在同时写变量冲突,在协程中控制共享资源不加锁,只需要判断状态就好了,所以执行效率比多线程高很多。

         我们以爬虫为例子,说明协程的应用。启动一个爬虫程序一定属于一个进程,这是毋庸置疑的,但进程本身并不会执行任何操作,所有操作都是通过线程来完成,所以一个进程有一个主线程。一般爬虫的步骤包括发送request请求、写入文件等操作,而这些都是IO操作,当线程执行到这些操作时,要么等待这一操作完成要么切换到其他线程。如果使用了协程呢?如果协程遇到了此类IO操作,可以立即切换到其他操作,例如直接发送下一个request请求,甚至发送第二个、第三个请求……直至原来的协程中IO请求完成,那么回到原来的协程继续下一步操作。这就是协程的工作原理,充分利用线程的工作效率,也没有多线程切换的开销,所以在处理IO操作时协程非常高效。

       简单总结一下协程的优缺点:

       优点:

  1)无需线程上下文切换的开销(还是单线程);

  2)无需原子操作的锁定和同步的开销;

  3)方便切换控制流,简化编程模型;

  4)高并发+高扩展+低成本:一个cpu支持上万的协程都没有问题,适合用于高并发处理。

  缺点:

  1)无法利用多核的资源,协程本身是个单线程,它不能同时将单个cpu的多核用上,协程需要和进程配合才能运用到多cpu上(协程是跑在线程上的);

  2)进行阻塞操作时会阻塞掉整个程序:如io;

       现在,各位读者应该已经对协程的概念又说了解了,也感受到了协程的魅力了吧!那么该怎么使用协程了……

2.3 相关概念

  在Python中, asyncio、tornado和gevent等模块都实现了协程的功能。本篇中主要介绍asyncio。

  在介绍通过asyncio的使用协程之前,首先有必要先介绍一下asyncio中涉及的几个概念,要想掌握asyncio,这几个贯穿始终的概念必须好好理解:

  1)event_loop事件循环:程序开启一个事件的循环,程序员会把一些函数(协程)注册到事件循环上。当满足事件发生的时候,调用相应的协程函数。

  2)coroutine 协程:协程对象,指一个使用async关键字定义的函数,它的调用不会立即执行函数,而是会返回一个协程对象。协程对象需要注册到事件循环,由事件循环调用。

  3)future 对象: 代表将来执行或没有执行的任务的结果。它和task上没有本质的区别

  4)task 任务:一个协程对象就是一个原生可以挂起的函数,任务则是对协程进一步封装,其中包含任务的各种状态。Task 对象是 Future 的子类,它将coroutine和Future联系在一起,将 coroutine 封装成一个 Future 对象。

  5)async/await 关键字:python3.5 用于定义协程的关键字,async定义一个协程,await用于挂起阻塞的异步调用接口,await就类似于说下面过程阻塞,暂时执行别的协程。await关键字添加了一个新的协程到循环里,而不需要明确地添加协程到这个事件循环里。

3 定义协程

  定义协程比定义进程或者线程还要简单,你只需要在普通函数定义时在“def”关键字前面加上一个“asyncio”,即可把普通函数定义为一个协程:

async def firstCorouctine(path=‘a.txt’):

    print(‘协程执行开始……’)

    await asyncio.sleep(1)

    print(‘协程执行结束……’)

  如果单单看firstCorouctine(),我想大家都看得出这个函数的功能。有 “async”这个关键字在前面修饰之后,这个函数就变成了一个协程——这就是定义协程的方法。所以说,协程某种意义上就是一个函数。我们可以通过asyncio.iscoroutinefunction函数来查看某个函数到底是不是协程,如果是协程则返回True,否则返回False:

import asyncio

async def firstCorouctine():

    print(‘协程执行开始……’)

    await asyncio.sleep(1)

    print(‘协程执行结束……’)

 

def fun(path):

    print(‘这是一个普通函数’)

 

print(‘firstCorouctine是协程吗:{}’.format(asyncio.iscoroutinefunction(firstCorouctine)))

print(‘fun是协程吗:{}’.format(asyncio.iscoroutinefunction(fun)))

  输出结果:

  firstCorouctine是协程吗:True

  fun是协程吗:False

怎么样,简单吧?不过我猜,你心里还是一团懵,甚至在想这有什么用呢?请继续往下看。

4 使用协程

4.1 单个协程

         使用协程须得经过一下几个步骤:定义协程->(封装成task->)获取事件循环->将task放到事件循环中执行。定义好的协程并不能直接使用,需要将其包装成为了一个任务(task对象),然后放到事件循环中才能被执行。所谓task对象是Future类的一个子类,保存了协程运行后的状态,用于未来获取协程的结果。在上面的步骤中,之所以在封装task这一个步骤上加上括号,是因为我们也可以选择直接将协程放到事件循环中,事件循环会自动帮我们完成这一操作。

所以,从定义好一个协程,到执行一个协程就有不同的方法:

  第一种,通过asyncio 提供的ensure_future()函数创建task,然后执行

import asyncio

async def firstCorouctine(): # 定义协程

  print(‘协程执行开始……’)

  await asyncio.sleep(1)

  print(‘协程执行结束……’)


coroutine = firstCorouctine() # 将协程赋值给coroutine

task = asyncio.ensure_future(coroutine) # 封装为task

loop = asyncio.get_event_loop() # 获取事件循环

loop.run_until_complete(task) # 执行

  第二种,直接通过事件循环的create_task方法创建task,然后执行:

import asyncio

async def firstCorouctine(): # 定义协程

    print(‘协程执行开始……’)

    await asyncio.sleep(1)

    print(‘协程执行结束……’)

 

coroutine = firstCorouctine() # 将协程赋值给coroutine

loop = asyncio.get_event_loop() # 获取事件循环

task = loop.create_task(coroutine) # 封装为task

loop.run_until_complete(task) # 执行

  第三种:直接将协程放到事件循环中执行。这种方法并不是说不用将协程封装为task,而是事件循环内部会自动帮我们完成这一步骤。

import asyncio

async def firstCorouctine(): # 定义协程

  print(‘协程执行开始……’)

  await asyncio.sleep(1)

  print(‘协程执行结束……’)


coroutine = firstCorouctine() # 将协程赋值给coroutine

loop = asyncio.get_event_loop() # 获取事件循环

loop.run_until_complete(coroutine) # 执行

  当然,无论是上述哪一种方法,最终都需要通过run_until_complete方法去执行我们定义好的协程。run_until_complete 是一个阻塞(blocking)调用,直到协程运行结束,它才返回。这一点从函数名中就可以看得出来。

4.2 多协程并发

  协程往往是多个一起应用在事件循环里的,将多个协程加入事件循环需要借助 asyncio.gather 函数或者asyncio.wait函数,两个函数功能极其相似,不同的是,gather接受的参数是多个协程,而wait接受的是一个协程列表。async.wait会返回两个值:done和pending,done为已完成的task,pending为超时未完成的task。而async.gather只返回已完成task。

         使用waiter函数:

import asyncio

async def firstCorouctine(n): # 定义协程

    print('协程{}开始执行……'.format(n))

    await asyncio.sleep(1)

    print('协程{}结束执行……'.format(n))

task_list = [

    asyncio.ensure_future(firstCorouctine(1)),

    asyncio.ensure_future(firstCorouctine(2)),

    asyncio.ensure_future(firstCorouctine(3))

]

loop = asyncio.get_event_loop()

loop.run_until_complete(asyncio.wait(task_list)) # 使用wait函数

  输出结果:

  协程1开始执行……

  协程2开始执行……

  协程3开始执行……

  协程1结束执行……

  协程2结束执行……

  协程3结束执行…… 

  使用gather函数:

import asyncio

async def firstCorouctine(n):  # 定义协程

    print('协程{}开始执行……'.format(n))

    await asyncio.sleep(1)

    print('协程{}结束执行……'.format(n))

 
task1 = asyncio.ensure_future(firstCorouctine(1))

task2 = asyncio.ensure_future(firstCorouctine(2))

task3 = asyncio.ensure_future(firstCorouctine(3))


tasks = asyncio.gather(task1, task2, task3)#如果协程有返回值时,最好赋值给一个tasks,方便回去返回结果

loop = asyncio.get_event_loop()

loop.run_until_complete(tasks)

  输出结果:

  协程1开始执行……

  协程2开始执行……

  协程3开始执行……

  协程1结束执行……

  协程2结束执行……

  协程3结束执行……

4.3 获取返回值

  前面的章节中说到,协程本质上来说也是一种函数,既然是函数就可以返回值。那么,协程执行完后,怎么获取它的返回值呢?task是future实例化对象,它封装有一个result()方法,通过task调用result()方法,可以获取协程的返回值:

import asyncio

async def firstCorouctine(): # 定义协程

    await asyncio.sleep(1)

    return ‘1234567890’

coroutine = firstCorouctine() # 将协程赋值给coroutine

task = asyncio.ensure_future(coroutine) # 封装为task

loop = asyncio.get_event_loop() # 获取事件循环

loop.run_until_complete(task) # 执行

 

return_value = task.result() # 获取协程返回值

print(‘协程返回的值为:{}’.format(return_value))

  输出结果:

  协程返回的值为:1234567890

  上面的例子是单个协程是获取返回值,如果多个协程呢?使用多个协程并发时,将多个task列表传入事件循环中执行,返回的task列表中的每一个task对象就包含了返回值:

import asyncio

async def firstCorouctine(n): # 定义协程

    print('协程{}开始执行……'.format(n))

    await asyncio.sleep(1)

    print('协程{}结束执行……'.format(n))

    return n

task_list = [

    asyncio.ensure_future(firstCorouctine(1)),

    asyncio.ensure_future(firstCorouctine(2)),

    asyncio.ensure_future(firstCorouctine(3))

]

loop = asyncio.get_event_loop()

loop.run_until_complete(asyncio.wait(task_list))

for task in task_list:

    print(task.result())

  输出结果:

  协程1开始执行……

  协程2开始执行……

  协程3开始执行……

  协程1结束执行……

  协程2结束执行……

  协程3结束执行……

  1

  2

  3

4.4 绑定回调函数

  在实际应用中,协程执行结束之后并不意味这整个任务就完成了,还需要运行其他函数,且其他函数也会用到协程的返回值,这就涉及到回调函数。协程中设置回调函数时需要将future对象(也就是我们创建的task)传入函数中,不过这个传参是自动完成的,所以回调函数必须至少设置一个形参:

import asyncio

async def firstCorouctine(): # 定义协程

    await asyncio.sleep(1)
return '1234567890' def callBack(future): # 定义一个回调函数 print('我是回调函数,协程返回值为:{}'.format(future.result())) coroutine = firstCorouctine() task = asyncio.ensure_future(coroutine) task.add_done_callback(callBack) # 绑定回调函数 loop = asyncio.get_event_loop() loop.run_until_complete(task)

  输出结果:

  我是回调函数,协程返回值为:1234567890

  如果还需要传入其他参数,就需要借助偏函数(functools.partial)来辅助使用了,这时候切记future对象是放在最后的:

import asyncio

import functools

async def firstCorouctine(): # 定义协程

    await asyncio.sleep(1)

    return '1234567890'

 

def callBack(value , future): # 定义一个回调函数

    print('我是回调函数,你输入的第一个参数为:{}'.format(value))

    print('我是回调函数,协程返回值为:{}'.format(future.result()))

coroutine = firstCorouctine()

task = asyncio.ensure_future(coroutine)

task.add_done_callback(functools.partial(callBack , 123)) # 绑定回调函数

loop = asyncio.get_event_loop()

loop.run_until_complete(task)

  输出结果:

  我是回调函数,你输入的第一个参数为:123

  我是回调函数,协程返回值为:1234567890

4.5 协程的嵌套使用

  事件循环执行协程时是通过run_until_complete方法,这个方法只接收一个协程或者future对象作为参数。在前面章节中,我们在介绍多协程并发操作时,用的是asyncio.wait函数和asyncio.gather函数,这两个函数本身也是一个协程,当接收多个协程作为参数时,实际上是在wait(或gather)协程里面执行了我们传入的多个协程,然后把结果返回。这就证明,协程是可以嵌套的。我们也可以通过wait和gather来写我们自己的嵌套协程。

         使用wait函数嵌套:

import asyncio

async def innerCorouctine(n): # 嵌套在里层的协程

    print('innerCorouctine-{}开始执行……'.format(n))

    await asyncio.sleep(1)

    print('innerCorouctine-{}结束执行……'.format(n))

    return n

async def outerCorouctine():

    print('outerCorouctine开始执行……')

    coroutine1 = innerCorouctine(1)

    coroutine2 = innerCorouctine(2)

    coroutine3 = innerCorouctine(3)

    tasks = [

        asyncio.ensure_future(coroutine1),

        asyncio.ensure_future(coroutine2),

        asyncio.ensure_future(coroutine3)

    ]

     dones, pendings = await asyncio.wait(tasks)

    for task in dones:

        print('协程返回值:{} '.format(task.result()))

    print('outerCorouctine结束行……')

 
loop = asyncio.get_event_loop()

loop.run_until_complete(outerCorouctine())

  输出结果:

  outerCorouctine开始执行……

  innerCorouctine-1开始执行……

  innerCorouctine-2开始执行……

  innerCorouctine-3开始执行……

  innerCorouctine-1结束执行……

  innerCorouctine-2结束执行……

  innerCorouctine-3结束执行……

  协程返回值:1

  协程返回值:3

  协程返回值:2

  outerCorouctine结束行……

  使用gather方法进行嵌套:

import asyncio

async def innerCorouctine(n): # 嵌套在里层的协程

    print('innerCorouctine-{}开始执行……'.format(n))

    await asyncio.sleep(1)

    print('innerCorouctine-{}结束执行……'.format(n))

    return n

async def outerCorouctine():

    print('outerCorouctine开始执行……')

    coroutine1 = innerCorouctine(1)

    coroutine2 = innerCorouctine(2)

    coroutine3 = innerCorouctine(3)


    tasks = [

        asyncio.ensure_future(coroutine1),

        asyncio.ensure_future(coroutine2),

        asyncio.ensure_future(coroutine3)

    ]

    tasks = await asyncio.gather(*tasks)

    for task in tasks:

        print('协程返回值:{} '.format(task))

    print('outerCorouctine结束行……')


loop = asyncio.get_event_loop()

loop.run_until_complete(outerCorouctine())

  输出结果:

  outerCorouctine开始执行……

  innerCorouctine-1开始执行……

  innerCorouctine-2开始执行……

  innerCorouctine-3开始执行……

  innerCorouctine-1结束执行……

  innerCorouctine-2结束执行……

  innerCorouctine-3结束执行……

  协程返回值:1

  协程返回值:2

  协程返回值:3

  outerCorouctine结束行……

  当然,也还有第三种方法进行嵌套,那就是使用run_until_complete函数:

import asyncio

async def innerCorouctine(n): # 嵌套在里层的协程

    print('innerCorouctine-{}开始执行……'.format(n))

    await asyncio.sleep(1)

    print('innerCorouctine-{}结束执行……'.format(n))

    return n

async def outerCorouctine():

    print('outerCorouctine开始执行……')

    coroutine1 = innerCorouctine(1)

    coroutine2 = innerCorouctine(2)

    coroutine3 = innerCorouctine(3)

    tasks = [

        asyncio.ensure_future(coroutine1),

        asyncio.ensure_future(coroutine2),

        asyncio.ensure_future(coroutine3)

    ]

    for task in asyncio.as_completed(tasks):#使用as_completed函数

        result = await task

        print('协程返回值: {}'.format(result))

loop = asyncio.get_event_loop()

loop.run_until_complete(outerCorouctine())

  输出结果:

  outerCorouctine开始执行……

  innerCorouctine-1开始执行……

  innerCorouctine-2开始执行……

  innerCorouctine-3开始执行……

  innerCorouctine-1结束执行……

  innerCorouctine-2结束执行……

  innerCorouctine-3结束执行……

  协程返回值: 1

  协程返回值: 2

  协程返回值: 3

5 总结

  本文介绍了协程概念、意义和单线程下协程基本使用方法,但对于多线程下如何使用协程并未涉及,后续再进行补充。

  参考资料:

  http://python.jobbole.com/87541/

  http://python.jobbole.com/87310/