Python 異步IO

異步IO

CPU的速度遠遠快於磁盤、網絡等IO。在一個線程中，CPU執行代碼的速度極快，然而，一旦遇到IO操做，如讀寫文件、發送網絡數據時，就須要等待IO操做完成，才能繼續進行下一步操做。這種狀況稱爲同步IO。
在IO操做的過程當中，當前線程被掛起，而其餘須要CPU執行的代碼就沒法被當前線程執行了。
由於一個IO操做就阻塞了當前線程，致使其餘代碼沒法執行，因此咱們必須使用多線程或者多進程來併發執行代碼，爲多個用戶服務。每一個用戶都會分配一個線程，若是遇到IO致使線程被掛起，其餘用戶的線程不受影響。
多線程和多進程的模型雖然解決了併發問題，可是系統不能無上限地增長線程。因爲系統切換線程的開銷也很大，因此，一旦線程數量過多，CPU的時間就花在線程切換上了，真正運行代碼的時間就少了，結果致使性能嚴重降低。
因爲咱們要解決的問題是CPU高速執行能力和IO設備的龜速嚴重不匹配，多線程和多進程只是解決這一問題的一種方法。
另外一種解決IO問題的方法是異步IO。當代碼須要執行一個耗時的IO操做時，它只發出IO指令，並不等待IO結果，而後就去執行其餘代碼了。一段時間後，當IO返回結果時，再通知CPU進行處理。
能夠想象若是按普通順序寫出的代碼其實是無法完成異步IO的，異步IO模型須要一個消息循環，在消息循環中，主線程不斷地重複「讀取消息-處理消息」這一過程。
在「發出IO請求」到收到「IO完成」的這段時間裏，同步IO模型下，主線程只能掛起，但異步IO模型下，主線程並無休息，而是在消息循環中繼續處理其餘消息。這樣，在異步IO模型下，一個線程就能夠同時處理多個IO請求，而且沒有切換線程的操做。對於大多數IO密集型的應用程序，使用異步IO將大大提高系統的多任務處理能力。

協程

協程，又稱微線程，纖程。英文名Coroutine。
子程序，或者稱爲函數，在全部語言中都是層級調用，好比A調用B，B在執行過程當中又調用了C，C執行完畢返回，B執行完畢返回，最後是A執行完畢。
因此子程序調用是經過棧實現的，一個線程就是執行一個子程序。
子程序調用老是一個入口，一次返回，調用順序是明確的。而協程的調用和子程序不一樣。
協程看上去也是子程序，但執行過程當中，在子程序內部可中斷，而後轉而執行別的子程序，在適當的時候再返回來接着執行。
注意，在一個子程序中中斷，去執行其餘子程序，不是函數調用，有點相似CPU的中斷。好比子程序A、B，可是在A中是沒有調用B的，因此協程的調用比函數調用理解起來要難一些。
看起來A、B的執行有點像多線程，但協程的特色在因而一個線程執行，那和多線程比，協程有何優點？
最大的優點就是協程極高的執行效率。由於子程序切換不是線程切換，而是由程序自身控制，所以，沒有線程切換的開銷，和多線程比，線程數量越多，協程的性能優點就越明顯。
第二大優點就是不須要多線程的鎖機制，由於只有一個線程，也不存在同時寫變量衝突，在協程中控制共享資源不加鎖，只須要判斷狀態就行了，因此執行效率比多線程高不少。
由於協程是一個線程執行，那怎麼利用多核CPU呢？最簡單的方法是多進程+協程，既充分利用多核，又充分發揮協程的高效率，可得到極高的性能。
Python對協程的支持是經過generator實現的。
生產者生產消息後，直接經過yield跳轉到消費者開始執行，待消費者執行完畢後，切換回生產者繼續生產，效率極高：

def consumer():
    r = ''
    while True:
        n = yield r
        if not n:
            return
        print('[CONSUMER] Consuming %s...' % n)
        r = '200 OK'

def produce(c):
    c.send(None)
    n = 0
    while n < 5:
        n = n + 1
        print('[PRODUCER] Producing %s...' % n)
        r = c.send(n)
        print('[PRODUCER] Consumer return: %s' % r)
    c.close()

c = consumer()
produce(c)
複製代碼

執行結果：

[PRODUCER] Producing 1...
[CONSUMER] Consuming 1...
[PRODUCER] Consumer return: 200 OK
[PRODUCER] Producing 2...
[CONSUMER] Consuming 2...
[PRODUCER] Consumer return: 200 OK
[PRODUCER] Producing 3...
[CONSUMER] Consuming 3...
[PRODUCER] Consumer return: 200 OK
[PRODUCER] Producing 4...
[CONSUMER] Consuming 4...
[PRODUCER] Consumer return: 200 OK
[PRODUCER] Producing 5...
[CONSUMER] Consuming 5...
[PRODUCER] Consumer return: 200 OK
複製代碼

注意到consumer函數是一個generator，把一個consumer傳入produce後：

1. 首先調用c.send(None)啓動生成器；
2. 而後，一旦生產了東西，經過c.send(n)切換到consumer執行；
3. consumer經過yield拿到消息，處理，又經過yield把結果傳回；
4. produce拿到consumer處理的結果，繼續生產下一條消息；
5. produce決定不生產了，經過c.close()關閉consumer，整個過程結束。

整個流程無鎖，由一個線程執行，produce和consumer協做完成任務，因此稱爲「協程」，而非線程的搶佔式多任務。

asyncio

Python實現異步IO很是簡單，asyncio是Python 3.4版本引入的標準庫，直接內置了對異步IO的支持。
asyncio的編程模型就是一個消息循環。咱們從asyncio模塊中直接獲取一個EventLoop的引用，而後把須要執行的協程扔到EventLoop中執行，就實現了異步IO。
用asyncio的異步網絡鏈接來獲取sina、sohu和163的網站首頁代碼以下：

import asyncio

@asyncio.coroutine
def wget(host):
    print('wget %s...' % host)
    connect = asyncio.open_connection(host, 80)
    reader, writer = yield from connect
    header = 'GET / HTTP/1.0\r\nHost: %s\r\n\r\n' % host
    writer.write(header.encode('utf-8'))
    yield from writer.drain()
    while True:
        line = yield from reader.readline()
        if line == b'\r\n':
            break
        print('%s header > %s' % (host, line.decode('utf-8').rstrip()))
    # Ignore the body, close the socket
    writer.close()

loop = asyncio.get_event_loop()
tasks = [wget(host) for host in ['www.sina.com.cn', 'www.sohu.com', 'www.163.com']]
loop.run_until_complete(asyncio.wait(tasks))
loop.close()
複製代碼

執行結果以下：

wget www.sohu.com...
wget www.sina.com.cn...
wget www.163.com...
(等待一段時間)
(打印出sohu的header)
www.sohu.com header > HTTP/1.1 200 OK
www.sohu.com header > Content-Type: text/html
...
(打印出sina的header)
www.sina.com.cn header > HTTP/1.1 200 OK
www.sina.com.cn header > Date: Wed, 20 May 2015 04:56:33 GMT
...
(打印出163的header)
www.163.com header > HTTP/1.0 302 Moved Temporarily
www.163.com header > Server: Cdn Cache Server V2.0
...
複製代碼

@asyncio.coroutine把一個generator標記爲coroutine類型，而後，咱們就把這個coroutine扔到EventLoop中執行。
yield from語法可讓咱們方便地調用另外一個generator。因此線程不會等待IO操做，而是直接中斷並執行下一個消息循環。當yield from返回時，線程就能夠從yield from拿到返回值，而後接着執行下一行語句。
在此期間，主線程並未等待，而是去執行EventLoop中其餘能夠執行的coroutine了，所以咱們用Task封裝的三個coroutine能夠實現由同一個線程併發執行。

async/await

爲了簡化並更好地標識異步IO，從Python 3.5開始引入了新的語法async和await，可讓coroutine的代碼更簡潔易讀。
使用新語法，只須要作兩步簡單的替換：

1. 把@asyncio.coroutine替換爲async；
2. 把yield from替換爲await。

用新語法從新編寫上一節的代碼以下：

import asyncio

async def wget(host):
    print('wget %s...' % host)
    connect = asyncio.open_connection(host, 80)
    reader, writer = await connect
    header = 'GET / HTTP/1.0\r\nHost: %s\r\n\r\n' % host
    writer.write(header.encode('utf-8'))
    await writer.drain()
    while True:
        line = await reader.readline()
        if line == b'\r\n':
            break
        print('%s header > %s' % (host, line.decode('utf-8').rstrip()))
    # Ignore the body, close the socket
    writer.close()

loop = asyncio.get_event_loop()
tasks = [wget(host) for host in ['www.sina.com.cn', 'www.sohu.com', 'www.163.com']]
loop.run_until_complete(asyncio.wait(tasks))
loop.close()
複製代碼

剩下的代碼保持不變。

aiohttp

asyncio實現了TCP、UDP、SSL等協議，aiohttp則是基於asyncio實現的HTTP框架。
安裝aiohttp：

pip install aiohttp
複製代碼

而後編寫一個HTTP服務器，分別處理如下URL：

• / - 首頁返回b'<h1>Index</h1>'；
• /hello/{name} - 根據URL參數返回文本hello, %s!。

代碼以下：

import asyncio

from aiohttp import web

async def index(request):
    await asyncio.sleep(0.5)
    return web.Response(body=b'<h1>Index</h1>')

async def hello(request):
    await asyncio.sleep(0.5)
    text = '<h1>hello, %s!</h1>' % request.match_info['name']
    return web.Response(body=text.encode('utf-8'))

async def init(loop):
    app = web.Application(loop=loop)
    app.router.add_route('GET', '/', index)
    app.router.add_route('GET', '/hello/{name}', hello)
    srv = await loop.create_server(app.make_handler(), '127.0.0.1', 8000)
    print('Server started at http://127.0.0.1:8000...')
    return srv

loop = asyncio.get_event_loop()
loop.run_until_complete(init(loop))
loop.run_forever()
複製代碼

注意aiohttp的初始化函數init()也是一個coroutine，loop.create_server()則利用asyncio建立TCP服務。 祝願早日打敗新冠病毒，武漢加油！中國加油！