Python並行程式設計之IO模型

2022-06-13 14:04:13

五種IO模型

為了更好地瞭解IO模型，我們需要事先回顧下：同步、非同步、阻塞、非阻塞

同步（synchronous） IO
非同步（asynchronous） IO
阻塞（blocking） IO
非阻塞（non-blocking）IO

五種I/O模型包括：阻塞I/O、非阻塞I/O、訊號驅動I/O(不常用)、I/O多路轉接、非同步I/O。其中，前四個被稱為同步I/O。

上五個模型的阻塞程度由低到高為：阻塞I/O > 非阻塞I/O > 多路轉接I/O > 訊號驅動I/O > 非同步I/O，因此他們的效率是由低到高的。

1、阻塞I/O模型

在linux中，預設情況下所有的socket都是blocking,除非特別指定，幾乎所有的I/O介面 ( 包括socket介面 ) 都是阻塞型的。

如果所面臨的可能同時出現的上千甚至上萬次的使用者端請求，“執行緒池”或“連線池”或許可以緩解部分壓力，但是不能解決所有問題。總之，多執行緒模型可以方便高效的解決小規模的服務請求，但面對大規模的服務請求，多執行緒模型也會遇到瓶頸，可以用非阻塞介面來嘗試解決這個問題。

2、非阻塞I/O模型

在非阻塞式I/O中，使用者程序其實是需要不斷的主動詢問kernel資料準備好了沒有。但是非阻塞I/O模型絕不被推薦。

非阻塞，不等待。比如建立socket對某個地址進行connect、獲取接收資料recv時預設都會等待（連線成功或接收到資料），才執行後續操作。
如果設定setblocking(False)，以上兩個過程就不再等待，但是會報BlockingIOError的錯誤，只要捕獲即可。

非同步，通知，執行完成之後自動執行回撥函數或自動執行某些操作（通知）。比如做爬蟲中向某個地址baidu。com傳送請求，當請求執行完成之後自執行回撥函數。

3、多路複用I/O模型(事件驅動)

基於事件迴圈的非同步非阻塞框架:如Twisted框架，scrapy框架(單執行緒完成並行)。

檢測多個socket是否已經發生變化（是否已經連線成功/是否已經獲取資料）(可讀/可寫)IO多路複用作用？

作業系統檢測socket是否發生變化，有三種模式：

select：最多1024個socket；迴圈去檢測。
poll：不限制監聽socket個數；迴圈去檢測（水平觸發）。
epoll：不限制監聽socket個數；回撥方式（邊緣觸發）。

Python模組：

select.select
select.epoll

基於IO多路複用+socket非阻塞,實現並行請求(一個執行緒100個請求)

import socket
# 建立socket
client = socket.socket()
# 將原來阻塞的位置變成非阻塞（報錯）
client.setblocking(False)

# 百度建立連線: 阻塞
try:
    # 執行了但報錯了
    client.connect(('www.baidu.com',80))
except BlockingIOError as e:
    pass

# 檢測到已經連線成功

# 問百度我要什麼？
client.sendall(b'GET /s?wd=alex HTTP/1.0rnhost:www.baidu.comrnrn')

# 我等著接收百度給我的回覆
chunk_list = []
while True:
    # 將原來阻塞的位置變成非阻塞（報錯）
    chunk = client.recv(8096) 
    if not chunk:
        break
    chunk_list.append(chunk)

body = b''.join(chunk_list)
print(body.decode('utf-8'))

selectors模組

#伺服器端
from socket import *
import selectors

sel=selectors.DefaultSelector()
def accept(server_fileobj,mask):
    conn,addr=server_fileobj.accept()
    sel.register(conn,selectors.EVENT_READ,read)

def read(conn,mask):
    try:
        data=conn.recv(1024)
        if not data:
            print('closing',conn)
            sel.unregister(conn)
            conn.close()
            return
        conn.send(data.upper()+b'_SB')
    except Exception:
        print('closing', conn)
        sel.unregister(conn)
        conn.close()



server_fileobj=socket(AF_INET,SOCK_STREAM)
server_fileobj.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)
server_fileobj.bind(('127.0.0.1',8088))
server_fileobj.listen(5)
server_fileobj.setblocking(False) #設定socket的介面為非阻塞
sel.register(server_fileobj,selectors.EVENT_READ,accept) #相當於網select的讀列表裡append了一個檔案控制程式碼
                                                         #server_fileobj,並且繫結了一個回撥函數accept

while True:
    events=sel.select() #檢測所有的fileobj，是否有完成wait data的
    for sel_obj,mask in events:
        callback=sel_obj.data #callback=accpet
        callback(sel_obj.fileobj,mask) #accpet(server_fileobj,1)

#使用者端
from socket import *
c=socket(AF_INET,SOCK_STREAM)
c.connect(('127.0.0.1',8088))

while True:
    msg=input('>>: ')
    if not msg:continue
    c.send(msg.encode('utf-8'))
    data=c.recv(1024)
    print(data.decode('utf-8'))

4、非同步I/O

asyncio是Python 3.4版本引入的標準庫，直接內建了對非同步IO的支援。

asyncio的程式設計模型就是一個訊息迴圈。我們從asyncio模組中直接獲取一個EventLoop的參照，然後把需要執行的協程扔到EventLoop中執行，就實現了非同步IO。

用asyncio實現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()
# 執行coroutine
loop.run_until_complete(hello())
loop.close()

@asyncio.coroutine把一個generator標記為coroutine型別，然後，我們就把這個coroutine扔到EventLoop中執行。

hello()會首先列印出Hello world!，然後，yield from語法可以讓我們方便地呼叫另一個generator。由於asyncio.sleep()也是一個coroutine，所以執行緒不會等待asyncio.sleep()，而是直接中斷並執行下一個訊息迴圈。當asyncio.sleep()返回時，執行緒就可以從yield from拿到返回值（此處是None），然後接著執行下一行語句。

把asyncio.sleep(1)看成是一個耗時1秒的IO操作，在此期間，主執行緒並未等待，而是去執行EventLoop中其他可以執行的coroutine了，因此可以實現並行執行。

我們用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()

觀察執行過程：

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是由同一個執行緒並行執行的。

如果把asyncio.sleep()換成真正的IO操作，則多個coroutine就可以由一個執行緒並行執行。

我們用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.0rnHost: %srnrn' % host
    writer.write(header.encode('utf-8'))
    yield from writer.drain()
    while True:
        line = yield from reader.readline()
        if line == b'rn':
            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
...

可見3個連線由一個執行緒通過coroutine並行完成。

async/await

用asyncio提供的@asyncio.coroutine可以把一個generator標記為coroutine型別，然後在coroutine內部用yield from呼叫另一個coroutine實現非同步操作。

為了簡化並更好地標識非同步IO，從Python 3.5開始引入了新的語法async和await，可以讓coroutine的程式碼更簡潔易讀。

請注意，async和await是針對coroutine的新語法，要使用新的語法，只需要做兩步簡單的替換：

把@asyncio.coroutine替換為async；
把yield from替換為await。

讓我們對比一下上一節的程式碼：

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

用新語法重新編寫如下：

async def hello():
    print("Hello world!")
    r = await asyncio.sleep(1)
    print("Hello again!")

剩下的程式碼保持不變。

小結

asyncio提供了完善的非同步IO支援；

非同步操作需要在coroutine中通過yield from完成；

多個coroutine可以封裝成一組Task然後並行執行。

到此這篇關於Python並行程式設計之IO模型的文章就介紹到這了。希望對大家的學習有所幫助，也希望大家多多支援it145.com。