流畅python学习笔记第十八章：使用asyncio包处理并发

Posted 2020-10-28 一张红枫叶

tags:

篇首语：本文由小常识网(cha138.com)小编为大家整理，主要介绍了流畅python学习笔记第十八章：使用asyncio包处理并发相关的知识，希望对你有一定的参考价值。

首先是线程与协程的对比。在文中作者通过一个实例分别采用线程实现和asynchio包实现来比较两者的差别。在多线程的样例中，会用到join的方法，下面来介绍下join方法的使用。

知识点一：
当一个进程启动之后，会默认产生一个主线程，因为线程是程序执行流的最小单元，当设置多线程时，主线程会创建多个子线程，在python中，默认情况下（其实就是setDaemon(False)），主线程执行完自己的任务以后，就退出了，此时子线程会继续执行自己的任务，直到自己的任务结束，例子如下：。

def run():

time.sleep(2)

print(\'当前线程名称是:%s\\n\' % threading.currentThread().name)

time.sleep(2)

if __name__=="__main__":

start_time=time.time()

print(\'这是主线程:%s\' % threading.current_thread().name)

thread_list=[]

for i in range(5):

t=threading.Thread(target=run)

thread_list.append(t)

for t in thread_list:

t.start()

print(\'主线程结束:%s\' % threading.current_thread().name)

print(\'一共用时:%f\' % float(time.time()-start_time))

运行结果：

我们的计时是针对主线程的计时，主线程结束，计时也随之结束，打印出主线程的用时

主线程的任务完成之后，主线程随之结束，子线程继续执行自己的任务，直到全部的子线程的任务全部结束，程序结束。

/usr/bin/python2.7 /home/zhf/py_prj/function_test/asy_try.py

这是主线程:MainThread

主线程结束:MainThread

一共用时:0.000893

当前线程名称是:Thread-4

当前线程名称是:Thread-1

当前线程名称是:Thread-2

当前线程名称是:Thread-3

当前线程名称是:Thread-5

知识点二：
当我们使用setDaemon(True)方法，设置子线程为守护线程时，主线程一旦执行结束，则全部线程全部被终止执行，可能出现的情况就是，子线程的任务还没有完全执行结束，就被迫停止，例子如下。设置t.setDaemon(True)

for t in thread_list:

t.setDaemon(True)

t.start()

结果如下：主线程结束后，所有的线程都结束

/usr/bin/python2.7 /home/zhf/py_prj/function_test/asy_try.py

这是主线程:MainThread

主线程结束:MainThread

一共用时:0.000569

知识点三：
此时join的作用就凸显出来了，join所完成的工作就是线程同步，即主线程任务结束之后，进入阻塞状态，一直等待其他的子线程执行结束之后，主线程在终止，代码中添加如下。

for t in thread_list:

t.join()

运行结果：线程运行完后主线程才退出。

/usr/bin/python2.7 /home/zhf/py_prj/function_test/asy_try.py

这是主线程:MainThread

当前线程名称是:Thread-1

当前线程名称是:Thread-2

当前线程名称是:Thread-4

当前线程名称是:Thread-3

当前线程名称是:Thread-5

主线程结束:MainThread

一共用时:4.008916

介绍完了线程的用法。现在来看用多线程实现书中的例子。这个例子主要是产生一个|\\-/构成的旋转指针。代码如下：

class Signal:

go=True

def spin(msg,signal):

write,flush=sys.stdout.write,sys.stdout.flush

for char in itertools.cycle(\'|/-\\\\\'):

status=char+\'\'+msg

write(status)

flush()

write(\'\\x08\'*len(status))

time.sleep(1)

if not signal.go:

break

write(\'\'*len(status)+\'\\0x8\'*len(status))

def slow_function():

time.sleep(3)

return 42

def supervisor():

result=0

signal=Signal()

spinner=threading.Thread(target=spin,args=(\'thinking!\',signal))

print(\'spinner object:\',spinner)

spinner.start()

result=slow_function()

signal.go=False

spinner.join()

return result

def main():

result=supervisor()

print(\'Answer:\',result)

if __name__=="__main__":

main()

运行结果如下：在这个例子中，没有直接关闭线程，而是通过signal.go的方式来退出。

接下来介绍asyncio. asyncio的编程模型就是一个消息循环。我们从asyncio模块中直接获取一个EventLoop的引用，然后把需要执行的协程扔到EventLoop中执行，就实现了异步。来看下几个基本的概念：

event_loop 事件循环：程序开启一个无限的循环，程序员会把一些函数注册到事件循环上。当满足事件发生的时候，调用相应的协程函数。
coroutine 协程：协程对象，指一个使用async关键字定义的函数，它的调用不会立即执行函数，而是会返回一个协程对象。协程对象需要注册到事件循环，由事件循环调用。
task 任务：一个协程对象就是一个原生可以挂起的函数，任务则是对协程进一步封装，其中包含任务的各种状态。
future：代表将来执行或没有执行的任务的结果。它和task上没有本质的区别
async/await 关键字：python3.5 用于定义协程的关键字，async定义一个协程，await用于挂起阻塞的异步调用接口

我们先不看书中用asychio改造上面的例子。先来看下asyncio的工作原理。

代码如下

@asyncio.coroutine

def hello():

print(\'hello world\')

r=yield from asyncio.sleep(1)

print(\'hello again\')

代码也可以写成如下的形式：两种不同的书写方式。

async def hello():

print(\'hello world\')

await asyncio.sleep(1)

print(\'hello again\')

if __name__=="__main__":

loop=asyncio.get_event_loop()

task=loop.create_task(hello())

print(datetime.datetime.now())

print(task)

loop.run_until_complete(task)

print(task)

print(datetime.datetime.now())

loop.close()

运行结果：

/usr/bin/python3.6 /home/zhf/py_prj/function_test/asy_try.py

2018-03-23 22:00:59.179893

hello world

hello again

<Task finished coro=<hello() done, defined at /home/zhf/py_prj/function_test/asy_try.py:46> result=None> None

2018-03-23 22:01:00.181498

协程对象不能直接运行，在注册事件循环的时候，其实是run_until_complete方法将协程包装成为了一个任务（task）对象。所谓task对象是Future类的子类。保存了协程运行后的状态，用于未来获取协程的结果。在通过loop.create_task(hello())的时候，任务其实是处于pending状态。在hello中通过asyncio.sleep(1)耗时一秒最后任务执行完后状态变为done.

asyncio.ensure_future(coroutine) 和 loop.create_task(coroutine)都可以创建一个task，run_until_complete的参数是一个futrue对象。当传入一个协程，其内部会自动封装成task，task是Future的子类

通过上面的例子，我们可以总结下asyncio的应用场景了。譬如说你有一个脚本向3个不同服务器请求数据。有时，谁知什么原因，发送给其中一个服务器的请求可能意外地执行了很长时间。想象一下从第二个服务器获取数据用了10秒钟。在你等待的时候，整个脚本实际上什么也没干。如果你可以写一个脚本可以不去等待第二个请求而是仅仅跳过它，然后开始执行第三个请求，然后回到第二个请求，执行之前离开的位置通过这种切换任务最小化了空转时间。这就是asyncio的主要应场景。这和前面将到的多线程是不一样的。多线程其实是多个任务并行处理。而在asyncio中其实只有一个时间轴。但是对于在这条时间轴上执行任务来说，又不是顺序式的触发。这种触发方式类似与中断。当执行了任务A的时候，产生一个中断，执行任务B，B任务执行完后回到A的中断处继续执行剩余的代码。代码中的yield from就可以想成是一个中断。因此异步代码都是在一个线程中执行的。整个流程可以用下图来表示：

从上图可知：

1.消息循环是在线程中执行

2.从队列中取得任务

3.每个任务在协程中执行下一步动作

4.如果在一个协程中调用另一个协程(await <coroutine_name>)，会触发上下文切换，挂起当前协程，并保存现场环境(变量，状态)，然后载入被调用协程

5.如果协程的执行到阻塞部分(阻塞I/O，Sleep)，当前协程会挂起，并将控制权返回到线程的消息循环中，然后消息循环继续从队列中执行下一个任务...以此类推

6.队列中的所有任务执行完毕后，消息循环返回第一个任务

接下来我们就来看下在多个任务运行的时候，asyncio是如何进行任务阻塞的。

@asyncio.coroutine

def do_work(x):

print(\'doing work\',x)

yield from asyncio.sleep(1)

return \'Done after {}s\'.format(x)

if __name__=="__main__":

start=time.time()

loop=asyncio.get_event_loop()

tasks=[asyncio.ensure_future(do_work(1)),asyncio.ensure_future(do_work(2)),asyncio.ensure_future(do_work(3))]

tasks1=[do_work(1),do_work(2),do_work(3)]

loop.run_until_complete(asyncio.wait(tasks))

loop.close()

end=time.time()

print("Total time:{}".format(end-start))

运行结果如下：在代码中有三个任务do_work(1),do_work(2),do_work(3).当在do_work中执行asyncio.sleep(1)的时候将会将选择权交给主循环，主循环会去队列中查找下一个任务继续执行。

/usr/bin/python3.6 /home/zhf/py_prj/function_test/asy_try.py

doing work 1

doing work 2

doing work 3

Total time:1.0022118091583252

可以看到在采用asyncio的方式下。总共的运行方式为1秒。如果采用顺序执行的方式，执行的时间将是1+2+3=6秒。可以看出asyncio的方式能够大大缩短时间。注意这里必须使用asyncio.sleep(1)而不能使用time.sleep(1)，使用time.sleep()的方式将会挂起什么都不做。

我们用多线程的方式来做个对比：

def run():

tasks=[]

start=time.time()

t1=threading.Thread(target=do_work_thread,args=(1,))

t2 = threading.Thread(target=do_work_thread, args=(2,))

t3 = threading.Thread(target=do_work_thread, args=(3,))

tasks.append(t1)

tasks.append(t2)

tasks.append(t3)

for t in tasks:

t.start()