Python 原生協程------asyncio

協程

　　在python3.5之前，寫成的實現都是經過生成器的yield from原理實現的， 這樣實現的缺點是代碼看起來會很亂，因而3.5版本以後python實現了原生的協程，而且引入了async和await兩個關鍵字用於支持協程。因而在用async定義的協程與python的生成器完全分開。

async def downloader(url):
    return 'bobby'

async def download_url(url):
    html = await downloader(url)
    return html
if __name__ == '__main__':
    coro = download_url('http://www/imooc.com')
    coro.send(None)

輸出結果：
Traceback (most recent call last):
  File "D:/MyCode/Cuiqingcai/Flask/test01.py", line 67, in <module>
    coro.send(None)
StopIteration: bobby

能夠看到結果中能夠將downloader(url)的結果返回。須要注意的是在原生協程裏面不能用next()來預激協程。

async def downloader(url):
    return 'bobby'

async def download_url(url):
    html = await downloader(url)
    return html
if __name__ == '__main__':
    coro = download_url('http://www/imooc.com')
    coro.next()

結果：
AttributeError: 'coroutine' object has no attribute 'next'
sys:1: RuntimeWarning: coroutine 'download_url' was never awaited

 

原生協程async代碼中間是不能在使用yield生成器的，這樣就爲了更好的將原生協程與生成器嚴格區分開來。而且await只能和async語句搭配，不能和生成器搭配。由於要調用await須要調用對象實現__await__()這個魔法方法。因此在定義協程時候注意不要混用。可是理解的時候仍是能夠將原生的協程中 await能夠對比生成器的yield from。

 asyncio

高併發的核心模塊，3.4以後引入，最具野性的模塊，web服務器，爬蟲均可以勝任。它是一個模塊也能夠看作一個框架。

協程編碼模式的三個要點：

1. 事件循環
2. 回調（驅動生成器（協程））
3. epoll（IO多路複用）

　　asyncio的簡單實用：

　　這裏須要注意的是：同步阻塞的接口不能使用在協程裏面。由於協程是單線程的，只要有一個地方阻塞了，那麼全部的協程都須要等待阻塞結束以後才能夠向下運行，因而在協程函數中等待必定不能用time.sleep() 若是用time.sleep()就失去了協程的意義了（即程序運行的時間將會是每一個協程數乘以time.sleep()的時間數。）同時用asyncio.sleep() 以前須要加上 await

import time
async def download_url(url): print('start get %s' % url) await asyncio.sleep(2) print('get %s finished.' % url) if __name__ == '__main__': start_time = time.time() loop = asyncio.get_event_loop() loop.run_until_complete(download_url('https:www.baidu.com')) # 阻塞等事件完成以後再向下運行。至關於進程線程的join()方法，或者進線程池的wait() print('一共用時：%s' % (time.time() - start_time)) start get https:www.baidu.com get https:www.baidu.com finished. 一共用時：2.0011143684387207

　　一次執行多個任務。（用時和一個任務同樣！）下面任務若是換成time.sleep(2)則函數須要至少20秒才能執行完畢。

import asyncio
import time async def download_url(url): print('start get %s.' % url) await asyncio.sleep(2) print('get %s end' % url) if __name__ == '__main__': start_time = time.time() loop = asyncio.get_event_loop() url_list = [] for i in range(10): url_list.append('https:www.baidu.com.index{}'.format(i)) tasks = [download_url(url) for url in url_list] loop.run_until_complete(asyncio.wait(tasks)) print('用時 %s' %(time.time() - start_time)) ... ... 用時 2.0031144618988037

　　獲取協程的返回值：

　　能夠用兩種方式先獲得一個future對象。而後將該future對象放入loop.run_until_complete()中。（該函數便可以接受future對象也能夠接受協程對象）而後future對象跟進程池和線程池中的future對象的方法是同樣的。因而能夠用.result()方法的到函數的返回結果。

async def download_url(url):
    # print('start get %s' % url)
    await asyncio.sleep(2) # print('get %s finished.' % url) return 'frank is a good man.' if __name__ == '__main__': start_time = time.time() loop = asyncio.get_event_loop() # get_future = asyncio.ensure_future(download_url('https:www.baidu.com')) get_task = loop.create_task(download_url('https:www.baidu.com')) loop.run_until_complete(download_url(get_task)) print('一共用時：%s' % (time.time() - start_time)) print(get_task.result()) 一共用時：2.0021142959594727 frank is a good man.

　　future完成以後的回調函數

　　設置協程完成以後的回調函數：future對象的.add_done_callback(),注意在調用add_done_callback的時候會默認將future對象傳遞給回調函數，所以回調函數必須至少接受一個參數，同時add_done_callback必須在run_until_complete()前，（協程函數建立以後調用）由於若是在run_until_complete()以後的話，協程都應結束了。就不會起做用了。

async def download_url(url):
    # print('start get %s' % url)
    await asyncio.sleep(2) # print('get %s finished.' % url) return 'frank is a good man.' def send_emai(future): print('網頁下載完畢') print(future.result()) if __name__ == '__main__': start_time = time.time() loop = asyncio.get_event_loop() # get_future = asyncio.ensure_future(download_url('https:www.baidu.com')) get_task = loop.create_task(download_url('https:www.baidu.com')) get_task.add_done_callback(send_emai) loop.run_until_complete(download_url(get_task)) print('一共用時：%s' % (time.time() - start_time)) print(get_task.result()) 網頁下載完畢 frank is a good man. 一共用時：2.0021145343780518 frank is a good man.

 

問題來了。tasks.add_done_callback方法只能接受函數名，若是回調的方法也須要參數怎麼辦？這就須要用到偏函數from functools import partial （偏函數能夠將函數包裝成爲另一個函數）

import  asyncio
import time
from functools import partial
async def get_html(url):
    print('start get url')
    await asyncio.sleep(2)
    return 'bobby'

def callback_method(url, future):# 此處由於是future對象即（tasks）調用的，因此有一個默認的參數。同時要注意：回調函數的future只能放到最後，其它的函數實參放前面。
    print('download ended %s' % url)

if __name__ == '__main__':
    start_time = time.time()
    loop = asyncio.get_event_loop()
    tasks = asyncio.ensure_future(get_html('http://www.baidu.com'))
    # tasks = loop.create_task(get_html('http://www.baidu.com'))
    tasks.add_done_callback(partial(callback_method, 'http://www.baidu.com')) # 參數爲回調方法
    loop.run_until_complete(tasks)
    print(tasks.result())


start get url
download ended http://www.baidu.com
bobby

 

gather 與 wait的區別。都是能夠等待程序運行以後往下運行。可是gather比wait更加高級一點

gather能夠將任務分組：

import  asyncio
import time

async def get_html(url):
    print('start get %s' % url)
    await asyncio.sleep(2)
    print('end get %s' % url)
if __name__ == '__main__':
    start_time = time.time()
    loop = asyncio.get_event_loop()
    group1 = [get_html('http://goup1.com') for i in range(2)]
    group2 = [get_html('http://group2.com') for i in range(2)]
    group1 = asyncio.gather(*group1)
    group2 = asyncio.gather(*group2)
    loop.run_until_complete(asyncio.gather(group1, group2))
    # 上面三行代碼也能夠合一塊兒loop.run_until_complete(asyncio.gather(*group1, *group2)) 注意此參數前面要加*
    print(time.time()-start_time)


start get http://goup1.com
start get http://group2.com
start get http://group2.com
start get http://goup1.com
end get http://goup1.com
end get http://group2.com
end get http://group2.com
end get http://goup1.com
2.0021145343780518

loop.run_forever()協程的任務完成以後不會中止，而是會一直運行。老師吐槽：python中間loop和future的關係有點亂，loop會被放到future中間同時future又能夠放到loop中間，形成一個循環。

如何取消future（task）

async def get_html(sleep_time):
    print('waiting')
    await asyncio.sleep(sleep_time)
    print('end after %s S' % sleep_time)

if __name__ == '__main__':
    task1 = get_html(2)
    task2 = get_html(3)
    task3 = get_html(4)
    tasks = [task1, task2, task3]
    loop = asyncio.get_event_loop()
    try:
        loop.run_until_complete(asyncio.wait(tasks))
    except KeyboardInterrupt as e:
        all_tasks = asyncio.Task.all_tasks()
        for task in all_tasks:
            print('cancel task')
            print(task.cancel())# 打印是否取消 是返回Ture,否False
        loop.stop()
        loop.run_forever() # 注意此處必定要加上loop.run_forever()否則會報異常 finally:
        loop.close()

waiting
waiting
waiting
cancel task
True
cancel task
True
cancel task
True
cancel task
True

 

將此代碼進入cmd中運行，而後再中間按ctrl + C鍵，主動生成一個 KeyboardInterrupt 異常，而後異常被捕捉以後作出處理（即中止協程的運行）

 

協程的嵌套

import asyncio
async def compute(x, y):
    print('Computer %s + %s...') % (x, y))
    await asyncio.sleep(1)
    return x + y

async def print_sum(x, y):
    result = await computer(x, y)
    print("%s + %s = %s" % (x, y, result))

loop = asyncio.get_event_loop()
loop.run_until_complete(print_sum(1, 2))
loop.close()

 

代碼分析圖：

 

1. loop首先會爲print_sum()建立一個task
2. event_loop()驅動task運行，使task進入（pending狀態）
3. task去執行print_sum
4. print_sum中首先進入子協程的調度（await至關於yiel from）因此轉向執行computer，print_sum自身暫停。
5. compute中存在await 因而也被迫進入暫停狀態，而後能夠直接返回給task（await == yield from 而yield from能夠在調度方與子生成器之間掠過委託方創建雙向通道）
6. task返回給event_loop()
7. 等待1秒鐘以後，task喚醒compute，compute繼續執行下一行代碼（即return x + y）完成以後compute就是一個done狀態，同時拋出一個stopiterationError異常，此異常將激活print_sum()(委託方)，而且將異常將被await（對應以前的yield from）捕捉並提取出return的值。
8. print_sum()被激活以後執行print而後變成done狀態，也會拋出一個stopiterationError異常，而後被task接受並處理了。

 asyncio中的其餘函數（如下三個爲底層函數，多數條件下用得很少）

call_soon()

　　在協程運行時候，能夠傳遞一些函數去執行，注意是函數不是協程。

import asyncio
def callback(sleep_times):
    print('sleep {} success'.format(sleep_times))

if __name__ =='__main__':
    loop = asyncio.get_event_loop()
    loop.call_soon(callback, 2) #第一個參數爲調用的函數名，第二個單數爲被調用函數的參數
    loop.run_forever()

sleep 2 success

 

　　注意此處調動函數的運行須要用到loop.run_forever()而不是loop.run_until_complete()由於oop.call_soon()只是調用函數而不是loop註冊的協程。

同時loop.run_forever()會致使程序一直在運行，不會自動結束。因而須要添加如下方法使程序關閉。

import asyncio
def callback(sleep_times):
    print('sleep {} success'.format(sleep_times))

def stop_loop(loop):
    loop.stop()

if __name__ =='__main__':
    loop = asyncio.get_event_loop()
    loop.call_soon(callback, 2)
    loop.call_soon(stop_loop, loop)
    loop.run_forever()

 

call_later()

　　功能，講一個callback函數在一個指定的時候運行。

import asyncio
def callback(sleep_times):
    print('sleep {} success'.format(sleep_times))

def stop_loop(loop):
    loop.stop()

if __name__ =='__main__':
    loop = asyncio.get_event_loop()
    loop.call_later(2, callback, 2) # 參數含義：第一個參數爲延遲時間，延遲越少越先運行。
    loop.call_later(1, callback, 1)
    loop.call_later(3, callback, 3)
    loop.call_soon(callback, 4)
    loop.run_forever()


sleep 4 success
sleep 1 success
sleep 2 success
sleep 3 success

 

同時存在call_soon()和call_later()時，call_soon()會在那個call_later()前面調用。

call_at()

　　也是調用函數在指定時間內調用，可是它的指定時間，是指的loop內的時間，而不是本身傳遞的時間。能夠用loop.time()來獲取loop內時間。

import asyncio
def callback(sleep_times):
    print('sleep {} success'.format(sleep_times))

if __name__ =='__main__':
    loop = asyncio.get_event_loop()
    now = loop.time()
    loop.call_at(now+2, callback, 2) # 注意第一個參數是相對於loop系統時間而來的，不是自定義的幾秒鐘以後運行。
    loop.call_at(now+1, callback, 1)
    loop.call_at(now+3, callback, 3)
    loop.call_soon(callback, 4)
    loop.run_forever()

sleep 4 success
sleep 1 success
sleep 2 success
sleep 3 success

 

協程中線程的實現

協程不提供阻塞的方式，可是有時候有些庫，和有些接口只能提供阻塞方式鏈接。因而就能夠在協程中繼承阻塞io