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

前面介绍了asyncio的用法。下面我们来看下如何用协程的方式来实现之前的旋转指针的方法

@asyncio.coroutine

def spin(msg):

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

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

        status=char+''+msg

        write(status)

        flush()

        write('x08'*len(status))

        try:

            yield from asyncio.sleep(0.1)

        except asyncio.CancelledError:

            break

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

 

@asyncio.coroutine

def slow_function():

    yield from asyncio.sleep(3)

    return 42

 

@asyncio.coroutine

def supervisor():

    spinner=asyncio.ensure_future(spin('thinking'))

#    spinner=asyncio.async(spin('thinking'))

    print('spinner object:',spinner)

    result=yield from slow_function()

    spinner.cancel()

return result

 

if __name__=="__main__":

    start=time.time()

    loop=asyncio.get_event_loop()

    result=loop.run_until_complete(supervisor())

    loop.close()

    print('Answer:',result)

    end=time.time()

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

运行的结果和之前用多线程是一样的。我们来看下运行的具体原理

1 asyncio.get_event_loop和 loop.run_until_complete(supervisor())

创建主循环，并传入任务supervisor

2 在supervisor中首先通过asyncio.async(spin('thinking'))将spin函数添加如Task。方法也可以是通过spinner=asyncio.ensure_future(spin('thinking'))。在asyncio中Task对象和threading.Thread的用途是一样的。Task对象用于驱动协程，Thread对象用于调用可调用的对象,Task对象不用自己实例化，而是通过把协程传给asyncio.async函数或者是asyncio.ensure_future，或者是loop.create_task

3 在spin函数中，执行在终端输出旋转指针。并通过asyncio.sleep(0.1)的方式让出控制权，会到主循环

4 此时来到supervisor函数。此时进入slow_function，在slow_function中asyncio.sleep(3)进行休眠，并在休眠期把控制权交给主循环。此时主循环会将控制权又交给spin函数。

5 3秒休眠结束后，返回42,并通过spinner.cancel函数取消spin的task,取消后会在协程当前暂停的yield出抛出asyncio.CancelledError异常。至此整个程序运行完毕。

我们继续来看下用asyncio来实现图片下载的程序

DEST_URL='downloads/'

BASE_URL1='http://seopic.699pic.com/photo/40011'

down_list=('8840.jpg_wh1200','7347.jpg_wh1200','6876.jpg_wh1200','6876.jpg_wh1200')

def save_flag(img,filename):

    path=os.path.join(DEST_URL,filename)

    with open(path,'wb') as f:

        f.write(img)

 

@asyncio.coroutine

def get_flag(cc):

    url='{}/{cc}.jpg'.format(BASE_URL1,cc=cc)

    print(url)

    resp = yield from aiohttp.ClientSession().get(url)

    print (resp.status)

    image = yield from resp.read()

    return image

 

def show(text):

    print(text)

    sys.stdout.flush()

 

@asyncio.coroutine

def download_one(cc):

    image = yield from get_flag(cc)

    show(cc)

    save_flag(image, cc + '.jpg')

    return cc

 

def download_many(cc_list):

    loop=asyncio.get_event_loop()

    to_do=[download_one(cc) for cc in sorted(cc_list)]

    wait_coro=asyncio.wait(to_do)

    res,_=loop.run_until_complete(wait_coro)

    loop.close()

    return len(res)

def main(download_many):

    t1=time.time()

    count=download_many(down_list)

    elapsed=time.time()-t1

    msg='n{} flags downloaded in {:.2f}s'

    print(msg.format(count,elapsed))

if __name__=="__main__":

main(download_many)

在这里我们用yield from aiohttp.ClientSession().get(url)

代替了request函数。因为request函数是个IO阻塞型的函数。注意aiohttp必须要安装才能使用。书上写的是用aiohttp.request(‘GET’,url)方法，但是我在实际使用的时候发现无法下载，提示如下错误：

client_session: <aiohttp.client.ClientSession object at 0x7fac75231f28>

Unclosed client session

client_session: <aiohttp.client.ClientSession object at 0x7fac75231dd8>

Unclosed client session

client_session: <aiohttp.client.ClientSession object at 0x7fac75231eb8>

Unclosed client session

client_session: <aiohttp.client.ClientSession object at 0x7fac75231f28>

Task exception was never retrieved

future: <Task finished coro=<download_one() done, defined at /home/zhf/py_prj/function_test/asy_try.py:51> exception=TypeError("'_SessionRequestContextManager' object is not iterable",)>

Traceback (most recent call last):

  File "/home/zhf/py_prj/function_test/asy_try.py", line 53, in download_one

    image = yield from get_flag(cc)

  File "/home/zhf/py_prj/function_test/asy_try.py", line 39, in get_flag

    resp=yield from aiohttp.request('GET',url)

TypeError: '_SessionRequestContextManager' object is not iterable

Task exception was never retrieved

future: <Task finished coro=<download_one() done, defined at /home/zhf/py_prj/function_test/asy_try.py:51> exception=TypeError("'_SessionRequestContextManager' object is not iterable",)>

在网上搜了下，推荐使用aiohttp.ClientSession().get进行下载。这个函数能保证相关的TCP资源能够得到释放，比如TCP链接

在这里download_one和get_flag都用到了协程, 意味着必须像协程那样驱动，这样才能把控制权交还给时间循环

asyncio.wait分别把各个协程包装进一个task对象，最后的结果是，wait处理的所有对象都通过某种方式变成Future类的实例，wait是协程函数，因此返回的是一个协程或生成器对象。为了驱动协程，我们把协程传给run_until_completed方法。

运行时间：

5 flags downloaded in 0.34s

 

下面继续来优化下这个下载程序。

首先将代码改动下，在保存图片的时候将大小放大1000倍

def save_flag(img,filename):

    path=os.path.join(DEST_URL,filename)

    with open(path,'wb') as f:

        f.write(img*1000)

5 flags downloaded in 13.46s

下载的图片大小如下所示：一张图片493M

那么耗时的时间呢.总共耗费了13.46秒的时间。速率降低了40多倍

5 flags downloaded in 13.46s

原因是什么呢。原因就在与save_flag是一个阻塞性的函数(f.write)。save_flag函数阻塞了客户代码与asyncio事件循环共用的唯一线程。因此保存文件的时候，整个程序都会冻结。那么解决办法就是使用run_in_executor方法。download_one代码修改如下：

@asyncio.coroutine

def download_one(cc):

    image = yield from get_flag(cc)

    show(cc)

    loop=asyncio.get_event_loop()

loop.run_in_executor(None,save_flag,image,cc+'.jpg')

return cc

修改之后总共耗时1.5秒

5 flags downloaded in 1.50s