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

Posted 一张红枫叶

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

<Task pending coro=<hello() running at /home/zhf/py_prj/function_test/asy_try.py:46>>

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)都可以创建一个taskrun_until_complete的参数是一个futrue对象。当传入一个协程,其内部会自动封装成tasktaskFuture的子类

 

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

从上图可知:

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

2.从队列中取得任务

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

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

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

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

    for t in tasks:

        t.join()

    end=time.time()

print("The time is :{}s".format(end-start))

运行结果:可以看到asyncio的运行时间和多线程是一样的。

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

doing work 1

doing work 2

doing work 3

Done after 1s

Done after 3s

Done after 2s

The time is :1.0017800331115723s

 

以上是关于流畅python学习笔记第十八章:使用asyncio包处理并发的主要内容,如果未能解决你的问题,请参考以下文章

第十八章 lamp架构

第十八章读书笔记

《linux内核设计与实现》读书笔记第十八章

Linux内核设计与实现读书笔记——第十八章

Go语言精修(尚硅谷笔记)第十七和十八章

“全栈2019”Java异常第十八章:Exception详解