併發編程之IO模型

時間 2019-11-11

標籤併發編程模型简体版

原文原文鏈接

主要內容：python

　　1、IO模型介紹linux

　　2、阻塞IO程序員

　　3、非阻塞IO編程

　　4、多路複用服務器

　　5、異步IO網絡

1️⃣ IO模型介紹多線程

　　1 何爲同步、異步、阻塞和非阻塞併發

　　　同步：　　　　　app

#所謂同步，就是在發出一個功能調用時，在沒有獲得結果以前，該調用就不會返回。按照這個定義，其實絕大多數函數都是同步調用。可是通常而言，咱們在說同步、異步的時候，特指那些須要其餘部件協做或者須要必定時間完成的任務。 #舉例： #1. multiprocessing.Pool下的apply #發起同步調用後，就在原地等着任務結束，根本不考慮任務是在計算仍是在io阻塞，總之就是一股腦地等任務結束

　　　　異步：異步

#異步的概念和同步相對。當一個異步功能調用發出後，調用者不能馬上獲得結果。當該異步功能完成後，經過狀態、通知或回調來通知調用者。若是異步功能用狀態來通知，那麼調用者就須要每隔必定時間檢查一次，效率就很低（有些初學多線程編程的人，總喜歡用一個循環去檢查某個變量的值，這實際上是一 種很嚴重的錯誤）。若是是使用通知的方式，效率則很高，由於異步功能幾乎不須要作額外的操做。至於回調函數，其實和通知沒太多區別。 #舉例： #1. multiprocessing.Pool().apply_async() #發起異步調用後，並不會等待任務結束才返回，相反，會當即獲取一個臨時結果（並非最終的結果，多是封裝好的一個對象）。

　　　　阻塞：

#阻塞調用是指調用結果返回以前，當前線程會被掛起（如遇到io操做）。函數只有在獲得結果以後纔會將阻塞的線程激活。有人也許會把阻塞調用和同步調用等同起來，實際上他是不一樣的。對於同步調用來講，不少時候當前線程仍是激活的，只是從邏輯上當前函數沒有返回而已。 #舉例： #1. 同步調用：apply一個累計1億次的任務，該調用會一直等待，直到任務返回結果爲止，但並未阻塞住（即使是被搶走cpu的執行權限，那也是處於就緒態）; #2. 阻塞調用：當socket工做在阻塞模式的時候，若是沒有數據的狀況下調用recv函數，則當前線程就會被掛起，直到有數據爲止。

　　　　非阻塞：

#非阻塞和阻塞的概念相對應，指在不能馬上獲得結果以前也會馬上返回，同時該函數不會阻塞當前線程。

　　小結：

#1. 同步與異步針對的是函數/任務的調用方式：同步就是當一個進程發起一個函數（任務）調用的時候，一直等到函數（任務）完成，而進程繼續處於激活狀態。而異步狀況下是當一個進程發起一個函數（任務）調用的時候，不會等函數返回，而是繼續往下執行當，函數返回的時候經過狀態、通知、事件等方式通知進程任務完成。

#2. 阻塞與非阻塞針對的是進程或線程：阻塞是當請求不能知足的時候就將進程掛起，而非阻塞則不會阻塞當前進程

　　二、IO模型分類

　　通常分爲五類：

* blocking IO # 阻塞IO * nonblocking IO # 非阻塞IO * IO multiplexing # 多路複用 * signal driven IO # 信號驅動IO * asynchronous IO # 異步IO # signal driven IO（信號驅動IO）在實際中並不經常使用，因此主要介紹其他四種IO Model。

　　IO發生時涉及的對象和步驟：

　　　以read爲例，它主要涉及兩個系統對象，一個調用這個IO的process \(or thread\)，另外一個就是系統內核\(kernel\)。

　　當一個read操做發生時，該操做會經歷兩個階段：

1）等待數據準備 (Waiting for the data to be ready) 2）將數據從內核拷貝到進程中(Copying the data from the kernel to the process)

2️⃣ 阻塞IO（blocking IO ）

　　在linux中，默認狀況下全部的socket都是blocking，一個典型的讀操做流程大概是這樣：

　　當用戶進程調用了recvfrom這個系統調用，kernel就開始了IO的第一個階段：準備數據。

　　對於network io來講，不少時候數據在一開始尚未到達（好比，尚未收到一個完整的UDP包），

　　這個時候kernel就要等待足夠的數據到來。　　　

　　　　而在用戶進程這邊，整個進程會被阻塞。當kernel一直等到數據準備好了，它就會將數據從kernel中拷貝到用戶內存， 　　而後kernel返回結果，用戶進程才解除block的狀態，從新運行起來。因此，blocking IO的特色就是在IO執行的兩個階段
　　（等待數據和拷貝數據兩個階段）都被block了。

幾乎全部的程序員第一次接觸到的網絡編程都是從listen\(\)、send\(\)、recv\(\) 等接口開始的， 使用這些接口能夠很方便的構建服務器/客戶機的模型。然而大部分的socket接口都是阻塞型的。以下圖 ps： 所謂阻塞型接口是指系統調用（通常是IO接口）不返回調用結果並讓當前線程一直阻塞 只有當該系統調用得到結果或者超時出錯時才返回。

　　　　實際上，除非特別指定，幾乎全部的IO接口 ( 包括socket接口 ) 都是阻塞型的。這給網絡編程帶來了一個很大的問題，

　　如在調用recv(1024)的同時，線程將被阻塞，在此期間，線程將沒法執行任何運算或響應任何的網絡請求。

　　　　一個簡單的解決方案是：　　

在服務器端使用多線程（或多進程）。多線程（或多進程）的目的是讓每一個鏈接都擁有獨立的線程（或進程），
這樣任何一個鏈接的阻塞都不會影響其餘的鏈接。

　　實例以下：

　　　　分兩部分，客戶端（client.py）和服務端（server.py）

　　　　client.py

#！/usr/bin/env python3 #-*- coding:utf-8 -*- # write by congcong
from socket import *

def talk(): client = socket(AF_INET,SOCK_STREAM) client.connect(('127.0.0.1',8806)) while True: mes = input('>>>:').strip() if not mes:continue client.send(mes.encode('utf-8')) data = client.recv(1024) print(data.decode('utf-8')) client.close() if __name__ == '__main__': talk()

View Code

　　　　server.py

#！/usr/bin/env python3 #-*- coding:utf-8 -*- # write by congcong

from socket import *
from threading import Thread,currentThread def talk(conn): while True: try: data = conn.recv(1024) if not data:break
            print(data.decode('utf-8')) conn.send(('%s hello'%currentThread().getName()).encode('utf-8')) except ConnectionResetError: break conn.close() def server(ip,port): server = socket(AF_INET,SOCK_STREAM) server.bind((ip,port)) server.listen(5) while True: print('staring...') conn,addres = server.accept() print(addres) t = Thread(target=talk,args=(conn,)) t.start() server.close() if __name__ == '__main__': server('127.0.0.1',8806)

View Code

　　但存在問題下述問題：　

   雖然實現了併發，即實際則迴避了阻塞的問題（並未解決），思路是：讓主線程接收客戶端的連接，而當每收到了一個連接，就新建一個線程，負責收發消息，互不影響，並無監測IO,每一個線程遇到阻塞IO時，仍然阻塞，但並不影響其餘線程，從而實現併發。但會隨着客戶端連接的增多，服務端開的線程則愈來愈多，浪費資源，而爲了不機器崩潰。而設置線程池（適應問題規模較小的狀況），限制併發數目，效率下降，但保證了機器健康運行。
 　因此，至此，單線程下的IO阻塞問題仍未解決。

3️⃣ 非阻塞IO（nonblocking IO）

　　Linux下，能夠經過設置socket使其變爲non-blocking。當對一個non-blocking socket執行讀操做時，流程是這個樣子：

　　對流程圖理解以下：

　　當用戶進程發出read操做時，若是kernel中的數據尚未準備好，那麼它並不會block用戶進程，而是馬上返回一個error。
從用戶進程角度講 ，它發起一個read操做後，並不須要等待，而是立刻就獲得了一個結果。用戶進程判斷結果是一個error時，
它就知道數據尚未準備好，因而用戶就能夠在本次到下次再發起read詢問的時間間隔內作其餘事情，或者直接再次發送read操做。
一旦kernel中的數據準備好了，而且又再次收到了用戶進程的system call，那麼它立刻就將數據拷貝到了用戶內存（這一階段仍然是阻塞的），而後返回。

也就是說非阻塞的recvform系統調用調用以後，進程並無被阻塞，內核立刻返回給進程，若是數據還沒準備好，此時會返回一個error。進程在返回以後，能夠乾點別的事情，而後再發起recvform系統調用。重複上面的過程，
循環往復的進行recvform系統調用。這個過程一般被稱之爲輪詢。輪詢檢查內核數據，直到數據準備好，再拷貝數據到進程，進行數據處理。須要注意，拷貝數據整個過程，進程仍然是屬於阻塞的狀態。

　　因此，在非阻塞式IO中，用戶進程實際上是須要不斷的主動詢問kernel數據準備好了沒有。

　　實例：

　　　　客戶端程序

#！/usr/bin/env python3 #-*- coding:utf-8 -*- # write by congcong

from socket import * client = socket(AF_INET,SOCK_STREAM) client.connect(('127.0.0.1',6666)) while True: msg = input('>>>:').strip() if not msg:continue client.send(msg.encode('utf-8')) data = client.recv(1024) print(data.decode('utf-8')) client.close()

View Code

　　　　服務端程序

#！/usr/bin/env python3 #-*- coding:utf-8 -*- # write by congcong

# 非阻塞IO --> 單進程下多線程速度極快，當前程序效率很高（一直佔用CPU），但影響其它程序的執行
''' 非阻塞的recv系統調用調用以後，進程並無被阻塞，內核立刻返回給進程，若是數據還沒準備好， 此時會返回一個error。進程在返回以後，能夠乾點別的事情，而後再發起recv系統調用。重複上面的過程， 循環往復的進行recv系統調用。這個過程一般被稱之爲輪詢。輪詢檢查內核數據，直到數據準備好，再拷貝數據到進程， 進行數據處理。須要注意，拷貝數據整個過程，進程仍然是屬於阻塞的狀態。 '''
from socket import * server = socket(AF_INET,SOCK_STREAM) server.bind(('127.0.0.1',6666)) server.listen(5) server.setblocking(False) # 不阻塞，默認是阻塞
rlist = [] # 存放連接
slist = [] # 存放消息
print('staring...') while True: try: conn,addres = server.accept() # IO堵塞
 rlist.append(conn) print(rlist) except BlockingIOError: # 未收到連接拋異常
        #print('數據未準備好！')
        # 收消息
        del_rlist = [] for conn in rlist: # 遍歷連接列表
            try: data = conn.recv(1024) if not data: continue
                #conn.send(data.upper())
 slist.append((conn,data)) except BlockingIOError:  # 碰到IO阻塞
                continue
            except Exception: conn.close() del_rlist.append(conn) # 發消息
        del_slist = [] for item in slist: try: conn = item[0] data = item[1] conn.send(data.upper()) # 可能會拋異常，即IO阻塞
                del_slist.append(item) # 沒拋異常，就將發成功的信息加到將要被刪除的隊列中
            except BlockingIOError: # 未發送成功
                continue
        for item in del_slist: # 遍歷列表，將其中存放的已發送成功的信息刪除
 slist.remove(item) for conn in del_rlist: # 遍歷列表，將未收到數據的連接刪除
 rlist.remove(conn) server.close()

View Code

　　優勢：

可以在等待任務完成的時間裏幹其餘活了（包括提交其餘任務，也就是 「後臺」 能夠有多個任務在「」同時「」執行）。

　　缺點：

1. 循環調用recv()將大幅度推高CPU佔用率；這也是咱們在代碼中留一句time.sleep(2)的緣由,不然在低配主機下極容易出現卡機狀況 2. 任務完成的響應延遲增大了，由於每過一段時間纔去輪詢一次read操做，而任務可能在兩次輪詢之間的任意時間完成。 這會致使總體數據吞吐量的下降。

注意：優勢難掩它的缺點，非阻塞IO模型毫不被推薦。

4️⃣ 多路複用（IO multiplexing）

　　多路複用即select/epoll，有些地方也稱這種IO方式爲事件驅動IO。

　　select/epoll的好處就在於單個process就能夠同時處理多個網絡鏈接的IO。它的基本原理就是select/epoll，

這個function會不斷的輪詢所負責的全部socket，當某個socket有數據到達了，就通知用戶進程。它的流程如圖：

當用戶進程調用了select，那麼整個進程會被block，而同時，kernel會「監視」全部select負責的socket，
當任何一個socket中的數據準備好了，select就會返回。這個時候用戶進程再調用read操做，將數據從kernel拷貝到用戶進程。  
這個圖和blocking IO的圖其實並無太大的不一樣，事實上還更差一些。由於這裏須要使用兩個系統調用\(select和recvfrom\)，
而blocking IO只調用了一個系統調用\(recvfrom\)。可是，用select的優點在於它能夠同時處理多個connection。

　　在多路複用模型中，對於每個socket，通常都設置成爲non-blocking，可是，如上圖所示，

整個用戶的process實際上是一直被block的。只不過process是被select這個函數block，而不是被socket IO給block。

　　注意：select的優點在於能夠處理多個鏈接，不適用於單個鏈接。

實例：

　　客戶端程序

#！/usr/bin/env python3 #-*- coding:utf-8 -*- # write by congcong
from socket import * client = socket(AF_INET,SOCK_STREAM) client.connect(('127.0.0.1',6868)) while True: mes = input('>>>:').strip() if not mes:continue client.send(mes.encode('utf-8')) data = client.recv(1024) print(data.decode('utf-8')) client.close()

View Code

　　服務端程序

#！/usr/bin/env python3 #-*- coding:utf-8 -*- # write by congcong

# 多路複用IO,能夠同時檢測多個IO,而非阻塞IO只能檢測一個IO,單個連接時非阻塞IO效率高，多個連接時多路複用更佳
''' select/epoll的好處就在於單個process就能夠同時處理多個網絡鏈接的IO。 它的基本原理就是select/epoll這個function會不斷的輪詢所負責的全部socket， 當某個socket有數據到達了，就通知用戶進程。 '''
from socket import *
import select server = socket(AF_INET,SOCK_STREAM) server.bind(('127.0.0.1',6868)) server.listen(5) server.setblocking(False) # 不阻塞
rlist = [server,] # 存放連接和套接字(server,conn)
wlist = [] # 存放發送消息的套接字()
wdata = { } while True: rl,wl,xl = select.select(rlist,wlist, [], 0.5)  # 後兩個參數分別表示異常列表和超時時間
    print(wl) # 收消息
    for sock in rl: if sock == server: # 收到server
            conn,addres = sock.accept() rlist.append(conn) # 加入列表
        else: # 即收到的是 conn
            try: data = sock.recv(1024) if not data: sock.close() rlist.remove(sock) # 針對linux系統報錯，一直接收的特色
 wlist.append(sock) wdata[sock] = data.upper() except Exception: # 關閉未收到數據的無用連接和刪除套接字
 sock.close() rlist.remove(sock) # 發消息
    for sock in wl: data = wdata[sock] # 獲取字典套接字對應的數據
        sock.send(data) # 發送數據
        wlist.remove(sock) # 刪除已經接收的套接字
        wdata.pop(sock) # 刪除已經發送成功的數據
server.close()

View Code

select監聽fd變化的過程分析：

用戶進程建立socket對象，拷貝監聽的fd到內核空間，每個fd會對應一張系統文件表，內核空間的fd響應到數據後，
就會發送信號給用戶進程數據已到；
用戶進程再發送系統調用，好比（accept）將內核空間的數據copy到用戶空間，同時做爲接受數據端內核空間的數據清除，
這樣從新監聽時fd再有新的數據又能夠響應到了（發送端由於基於TCP協議因此須要收到應答後纔會清除）。

該模型的優勢：

相比其餘模型，使用select() 的事件驅動模型只用單線程（進程）執行，佔用資源少，不消耗太多 CPU，同時可以爲多客戶端提供服務。
若是試圖創建一個簡單的事件驅動的服務器程序，這個模型有必定的參考價值。

該模型的缺點：

首先select()接口並非實現「事件驅動」的最好選擇。由於當須要探測的句柄值較大時，select()接口自己須要消耗大量時間去輪詢各個句柄。 不少操做系統提供了更爲高效的接口，如linux提供了epoll，BSD提供了kqueue，Solaris提供了/dev/poll，…。 若是須要實現更高效的服務器程序，相似epoll這樣的接口更被推薦。遺憾的是不一樣的操做系統特供的epoll接口有很大差別， 因此使用相似於epoll的接口實現具備較好跨平臺能力的服務器會比較困難。 其次，該模型將事件探測和事件響應夾雜在一塊兒，一旦事件響應的執行體龐大，則對整個模型是災難性的。

5️⃣ 異步IO

　　Linux下的asynchronous IO其實用得很少，從內核2.6版本纔開始引入。它的流程以下：

原理分析：

　　用戶進程發起read操做以後，馬上就能夠開始去作其它的事。而另外一方面，從kernel的角度，

當它受到一個asynchronous read以後，首先它會馬上返回，因此不會對用戶進程產生任何block。而後，

kernel會等待數據準備完成，而後將數據拷貝到用戶內存，當這一切都完成以後，kernel會給用戶進程

發送一個signal，告訴它read操做完成了。