Python說文解字_Python之多任務_05

時間 2019-11-25

原文原文鏈接

問：在Py3.5以前yield表現很是好，在Py3.5以後爲了將予以變得更加明確，就引入了async和await關鍵詞用於定義原生的協議。html

答：async和await原生協程：nginx

async def downloader(url):
    return "bobby"

async def downloader_url(url):
    # do something
    html = await downloader(url)
    return html

if __name__ == '__main__':
    coro = downloader_url("http://www.baidu.com")
    next(None)
    # coro.send(None)

# 1.若是調用send正常
# StopIteration: bobby
# 2.若是調用next草異常，所以原生協程只能用send
# TypeError: 'NoneType' object is not an iterator
# sys:1: RuntimeWarning: coroutine 'downloader_url' was never awaited

　　咱們發現：原生協程只能用send不能用next。並且發現原生協程和yield協程差很少，前面加上了async語法，await相似於yield from。Python引入了async和await原生協程是爲了咱們的語義更加的清晰。若是咱們用生成器寫出的協程的話，代碼很是的凌亂的。由於它又能當生成器又能當協程，顯得比較凌亂，將這兩種區分開來。所以async裏面是不能定義yield的。所以Python增強了咱們的區別。所以這兩個是一對的。這樣咱們的協程區分開來。前面說了那麼多生成器就是爲了增強協程的理解。這樣咱們在協程裏面就用這兩個。所以在Python內部依然沿用了生成器的原理，來實現了咱們的協程。程序員

　　await後跟隨的Awaitbale對象。咱們能夠經過from collections import Awaitalbe模塊。web

　　其實這個是實現了魔法拿書中的__await__的方法，所以咱們還可使用裝飾器的方法來操做，省去asyn，而變換成咱們熟悉的生成器的樣子。代碼以下：django

import types

@types.coroutine
def downloader(url):
    yield "bobby"

async def downloader_url(url):
    # do something
    html = await downloader(url)
    return html

if __name__ == '__main__':
    coro = downloader_url("http://www.baidu.com")
    # next(None)
    coro.send(None)

問：生成器是如何變成咱們協程的？編程

答：在開始我麼引入過協程的需求，咱們的協程是經過單線程調度，協程是咱們函數級別的是由咱們程序員本身來決定調用的，咱們能夠寫同步代碼同樣寫異步代碼。咱們的生成器就能夠完成咱們的協程的這麼一個功能。咱們如今就能夠用協程來模擬咱們的需求。flask

　　生成器是能夠暫停的函數，實際上生成器是能夠有狀態的！服務器

　　咱們看這段代碼多線程

import inspect
def gen_func():
    yield 1
    return "bobby"

if __name__ == '__main__':
    gen = gen_func()
    print(inspect.getgeneratorstate(gen)) # GEN_CREATED
    next(gen)
    print(inspect.getgeneratorstate(gen)) # GEN_SUSPENDED
    try:
        next(gen)
    except StopIteration:
        pass
    print(inspect.getgeneratorstate(gen)) # GEN_CLOSED

　　經過inspect中的getgeneratorstate咱們來觀察生成器的狀態，實際上咱們在定義咱們的生成器的時候，生成器能夠接收咱們的值。這句話有兩個意思：第一是返回值給調用方，第二調用方經過send方式返回值跟gen。如今咱們生成器由「生產者」變爲「消費者」。
併發

　　1.咱們用同步的方式編寫異步的代碼。

　　2.在適當的時候暫停函數，並在適當的時候啓動函數。

　　如今咱們模式：事件循環+協程模式。

　　咱們在函數當中的子函數，若是出現異常，會拋給這個函數的主函數，是「向上拋」的過程。這個就很好。協程是一個單線程模式。

問：異步IO和IO複用，也就是同步IO和異步IO。

答：咱們對前面的東西略微作一個小結：

　　異步IO和協程：如今咱們尚未把協程來用到咱們的編碼當中，協程是須要事件循環來實現的。單獨使用的話做用不是很明顯。

　　在最開始的時候我麼說到了併發、並行、異步、同步、阻塞、非阻塞。

　　在IO多路複用（同步IO）當中的select poll epoll，使咱們使用的最多的技術。回調+事件循環的方式。這種編程模式和同步IO的編程模式差異很大。

　　所以這兩種模式：回調+事件循環（IO多路複用）、協程+事件循環（異步IO）

　　上面的編碼是很是痛苦的：回調之痛。

　　咱們引入了生成器和協程，協程並不會別上面的方式高，協程主要解決的問題是回調之痛的問題和編碼習慣的問題。

　　咱們能夠將生成器編程咱們的協程了。

　　最後引入了async和await來區別生成器和協程，不容易混亂，進行區分。咱們能夠用Cororoutine裝飾器的方式，就不要用了。

　　因此建議使用async和await的方式。

問：async IO併發編程：

答：該模塊是在Python3.4後引入的模塊，這是Python編程中最難的部分。該模塊也是Python最具野心的模塊。分幾個部分開始講解：

　　1. 事件循環：

　　咱們能夠把async IO看作一個模塊也可看作一個框架，它完成了整套異步編程中最核心的內容。它包含各類特定系統實現的模塊化事件循環，傳輸和協議抽象；對TCP,UDP,SSL，子進程，延時調用以及其餘的具體支持；模仿futures模塊但適用於事件循環使用Future類；基於yield from的協議和任務，可讓你用順序的方式編寫併發代碼；必須使用一個將產生阻塞IO的調用時，有接口能夠把這個事件轉移到線程池。能夠將多進程和多線程協調進來。

　　協程編碼模式都逃離不掉三個要素：事件循環+調用（驅動生成器）+epoll（IO多路複用）

　　asyncio 是Python用於解決異步IO編程的一整套解決方案。

　　tornado、gevent、twisted（scrapy，django channels）

　　tornado：實現了web服務器，djago+flask是Python最傳統要搭配（uwsgi，gunicorn+nginx），tornado能夠直接部署，nginx+tornado

　　使用asyncio

import asyncio # 能夠當作協程池來理解比較容易

import time


async def get_html(url):
    print("start get url")
    # time.sleep(2) # 阻塞式的IO不能寫在裏面
    await asyncio.sleep(2) # 不能使用import time,必需要加await
    print("end get url")

if __name__ == '__main__':
    start_time = time.time()
    loop = asyncio.get_event_loop()
    loop.run_until_complete(get_html("htttp://www.baidu.com"))
    print(time.time() - start_time)

# start get url
# end get url
# 2.0150375366210938

　　get_event_loop市價循環

　　run_until_complete去執行

　　這裏不能用time.sleep這是阻塞式的方法。所以會單獨的一個一個執行很是慢，因此要使用asynic中的sleep

import asyncio # 能夠當作協程池來理解比較容易

import time


async def get_html(url):
    print("start get url")
    # time.sleep(2) # 阻塞式的IO不能寫在裏面
    await asyncio.sleep(2) # 不能使用import time,必需要加await
    print("end get url")

if __name__ == '__main__':
    start_time = time.time()
    loop = asyncio.get_event_loop()
    tasks = [get_html("htttp://www.baidu.com") for i in range(100)]
    loop.run_until_complete(asyncio.wait(tasks))
    print(time.time() - start_time)

# start get url
# end get url
# 2.0150375366210938
# time編程順序執行。asyncio.sleep()能夠當即執行。只要一個地方阻塞了其餘方面都實現不了。

　　咱們發現更改後就會阻塞。

import asyncio # 能夠當作協程池來理解比較容易

import time


async def get_html(url):
    print("start get url")
    # time.sleep(2) # 阻塞式的IO不能寫在裏面
    await asyncio.sleep(2) # 不能使用import time,必需要加await
    return "bobby"
if __name__ == '__main__':
    start_time = time.time()
    loop = asyncio.get_event_loop()
    # get_future = asyncio.ensure_future(get_html("htttp://www.baidu.com"))
    # loop.create_task()
    # tasks = [get_html("htttp://www.baidu.com") for i in range(100)]
    task = loop.create_task(get_html("htttp://www.baidu.com"))
    loop.run_until_complete(task)
    print(time.time() - start_time)
    print(task.result())


# 獲取協程的返回值

　　咱們用協程調用線程池：ensure_funture

　　使用方法還有create_task這兩種都是比較好理解的。

import asyncio # 能夠當作協程池來理解比較容易

import time
from functools import partial

async def get_html(url):
    print("start get url")
    # time.sleep(2) # 阻塞式的IO不能寫在裏面
    await asyncio.sleep(2) # 不能使用import time,必需要加await
    return "bobby"

def callback(url,future):
    print("send email to bobby")

if __name__ == '__main__':
    start_time = time.time()
    loop = asyncio.get_event_loop()
    # get_future = asyncio.ensure_future(get_html("htttp://www.baidu.com"))
    # loop.create_task()
    # tasks = [get_html("htttp://www.baidu.com") for i in range(100)]
    task = loop.create_task(get_html("htttp://www.baidu.com"))
    # task.add_done_callback(callback)
    task.add_done_callback(partial(callback,"htttp://www.baidu.com"))
    loop.run_until_complete(task)
    print(time.time() - start_time)
    print(task.result())


# 獲取協程的返回值

　　咱們也可使用回調，在task中的重寫add_done_callback方法。

import asyncio # 能夠當作協程池來理解比較容易

import time

async def get_html(url):
    print("start get url")
    # time.sleep(2) # 阻塞式的IO不能寫在裏面
    await asyncio.sleep(2) # 不能使用import time,必需要加await
    print("end get url")

if __name__ == '__main__':
    start_time = time.time()
    loop = asyncio.get_event_loop()
    tasks = [get_html("htttp://www.baidu.com") for i in range(100)]
    # loop.run_until_complete(asyncio.wait(tasks))
    loop.run_until_complete(asyncio.gather(*tasks))
    print(time.time() - start_time)

# wait 和 gather 的區別
# gather更加高層，能夠將咱們task分組
    group1 = [get_html("htttp://www.baidu1.com") for i in range(100)]
    group2 = [get_html("htttp://www.baidu2.com") for i in range(100)]
    loop.run_until_complete(asyncio.gather(*group1,*group2))

    group1 = asyncio.gather(*group1)
    group2 = asyncio.gather(*group2)

    group2.cancel()

　　咱們儘可能使用gather方法，注意他是能夠將咱們task進行分組，後面要加上*參數的形式。

　　2. task取消、嵌套、字寫成調用原理

# import asyncio
#
# loop = asyncio.get_event_loop()
# loop.run_forever()
# loop.run_until_complete()
# 1.loop會被放到future中。
# 2.取消future(task)

import asyncio
import time

async def get_html(sleep_times):
    print("waiting")
    await asyncio.sleep(sleep_times)
    print("done after {}s".format(sleep_times))

if __name__ == '__main__':
    task1 = get_html(2)
    task2 = get_html(3)
    task3 = get_html(2)
    tasks = [task1,task2,task3]

    loop = asyncio.get_event_loop()

    try:
        loop.run_until_complete(asyncio.wait(tasks))
    except KeyboardInterrupt as e:
        all_task = asyncio.Task.all_tasks()
        for task in all_task:
            print("cancel task")
            task.cancel()
        loop.stop()
        loop.run_forever()
    finally:
        loop.close()