Python使用asyncio標準庫對異步IO的支持

更新時間：2023年11月04日 15:19:51 投稿：yin

Python中,所有程序的執(zhí)行都是單線程的,但可同時執(zhí)行多個任務,不同的任務被時間循環(huán)（Event Loop）控制及調度,Asyncio是Python并發(fā)編程的一種實現(xiàn)方式；是Python 3.4版本引入的標準庫,直接內置了對異步IO的支持

Python中,所有程序的執(zhí)行都是單線程的,但可同時執(zhí)行多個任務,不同的任務被時間循環(huán)（Event Loop）控制及調度,Asyncio是Python并發(fā)編程的一種實現(xiàn)方式；是Python 3.4版本引入的標準庫,直接內置了對異步IO的支持。

異步IO

CPU的速度遠遠快于磁盤、網(wǎng)絡等IO。在一個線程中，CPU執(zhí)行代碼的速度極快，然而，一旦遇到IO操作，如讀寫文件、發(fā)送網(wǎng)絡數(shù)據(jù)時，就需要等待IO操作完成，才能繼續(xù)進行下一步操作。這種情況稱為同步IO。

在IO操作的過程中，當前線程被掛起，而其他需要CPU執(zhí)行的代碼就無法被當前線程執(zhí)行了。

因為一個IO操作就阻塞了當前線程，導致其他代碼無法執(zhí)行，所以我們必須使用多線程或者多進程來并發(fā)執(zhí)行代碼，為多個用戶服務。每個用戶都會分配一個線程，如果遇到IO導致線程被掛起，其他用戶的線程不受影響。

多線程和多進程的模型雖然解決了并發(fā)問題，但是系統(tǒng)不能無上限地增加線程。由于系統(tǒng)切換線程的開銷也很大，所以，一旦線程數(shù)量過多，CPU的時間就花在線程切換上了，真正運行代碼的時間就少了，結果導致性能嚴重下降。

由于我們要解決的問題是CPU高速執(zhí)行能力和IO設備的龜速嚴重不匹配，多線程和多進程只是解決這一問題的一種方法。

另一種解決IO問題的方法是異步IO。當代碼需要執(zhí)行一個耗時的IO操作時，它只發(fā)出IO指令，并不等待IO結果，然后就去執(zhí)行其他代碼了。一段時間后，當IO返回結果時，再通知CPU進行處理。

asyncio

asyncio是Python 3.4版本引入的標準庫，直接內置了對異步IO的支持。

asyncio的編程模型就是一個消息循環(huán)。我們從asyncio模塊中直接獲取一個EventLoop的引用，然后把需要執(zhí)行的協(xié)程扔到EventLoop中執(zhí)行，就實現(xiàn)了異步IO。

用asyncio實現(xiàn)Hello world代碼如下：

import asyncio

@asyncio.coroutine
def hello():
    print("Hello world!")
    # 異步調用asyncio.sleep(1):
    r = yield from asyncio.sleep(1)
    print("Hello again!")

# 獲取EventLoop:
loop = asyncio.get_event_loop()
# 執(zhí)行coroutine
loop.run_until_complete(hello())
loop.close()

@asyncio.coroutine把一個generator標記為coroutine類型，然后，我們就把這個coroutine扔到EventLoop中執(zhí)行。

hello()會首先打印出Hello world!，然后，yield from語法可以讓我們方便地調用另一個generator。由于asyncio.sleep()也是一個coroutine，所以線程不會等待asyncio.sleep()，而是直接中斷并執(zhí)行下一個消息循環(huán)。當asyncio.sleep()返回時，線程就可以從yield from拿到返回值（此處是None），然后接著執(zhí)行下一行語句。

把asyncio.sleep(1)看成是一個耗時1秒的IO操作，在此期間，主線程并未等待，而是去執(zhí)行EventLoop中其他可以執(zhí)行的coroutine了，因此可以實現(xiàn)并發(fā)執(zhí)行。

我們用Task封裝兩個coroutine試試：

import threading
import asyncio

@asyncio.coroutine
def hello():
    print('Hello world! (%s)' % threading.currentThread())
    yield from asyncio.sleep(1)
    print('Hello again! (%s)' % threading.currentThread())

loop = asyncio.get_event_loop()
tasks = [hello(), hello()]
loop.run_until_complete(asyncio.wait(tasks))
loop.close()

觀察執(zhí)行過程：

Hello world! (<_MainThread(MainThread, started 140735195337472)>)
Hello world! (<_MainThread(MainThread, started 140735195337472)>)
(暫停約1秒)
Hello again! (<_MainThread(MainThread, started 140735195337472)>)
Hello again! (<_MainThread(MainThread, started 140735195337472)>)

由打印的當前線程名稱可以看出，兩個coroutine是由同一個線程并發(fā)執(zhí)行的。

如果把asyncio.sleep()換成真正的IO操作，則多個coroutine就可以由一個線程并發(fā)執(zhí)行。

我們用asyncio的異步網(wǎng)絡連接來獲取sina、sohu和163的網(wǎng)站首頁：

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

執(zhí)行結果如下：

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
...

可見3個連接由一個線程通過coroutine并發(fā)完成。