day035協程、IO多路複用

 

本節內容：

一、協程（重點：gevent)
二、IO多路複用

1、協程

一、引子

本節的主題是基於單線程來實現併發，即只用一個主線程（很明顯可利用的cpu只有一個）狀況下實現併發，
爲此咱們須要先回顧下併發的本質：切換+保存狀態

　　cpu正在運行一個任務，會在兩種狀況下切走去執行其餘的任務（切換由操做系統強制控制），
一種狀況是該任務發生了阻塞，另一種狀況是該任務計算的時間過長或有一個優先級更高的程序替代了它

　　協程本質上就是一個線程，之前線程任務的切換是由操做系統控制的，遇到I/O自動切換，
如今咱們用協程的目的就是較少操做系統切換的開銷（開關線程，建立寄存器、堆棧等，在他們之間進行切換等），
在咱們本身的程序裏面來控制任務的切換。

　　ps：在介紹進程理論時，說起進程的三種執行狀態，而線程纔是執行單位，因此也能夠將上圖理解爲線程的三種狀態

一：其中第二種狀況並不能提高效率，只是爲了讓cpu可以雨露均沾，實現看起來全部任務都被「同時」執行的效果，

若是多個任務都是純計算的，這種切換反而會下降效率。
爲此咱們能夠基於yield來驗證。
yield自己就是一種在單線程下能夠保存任務運行狀態的方法，咱們來簡單複習一下：

#1 yiled能夠保存狀態，yield的狀態保存與操做系統的保存線程狀態很像，可是yield是代碼級別控制的，更輕量級
#2 send能夠把一個函數的結果傳給另一個函數，以此實現單線程內程序之間的切換

經過yield實現任務切換+保存狀態

import time def func1(): for i in range(11): #yield print('這是我第%s次打印啦' % i) time.sleep(1) def func2(): g = func1() #next(g) for k in range(10): print('哈哈，我第%s次打印了' % k) time.sleep(1) #next(g) #不寫yield，下面兩個任務是執行完func1裏面全部的程序纔會執行func2裏面的程序，有了yield，咱們實現了兩個任務的切換+保存狀態 func1() func2() 

Python

Copy

單純的切換反而會下降運行效率

#基於yield併發執行，多任務之間來回切換，這就是個簡單的協程的體現，可是他可以節省I/O時間嗎？不能 import time def consumer(): '''任務1:接收數據,處理數據''' while True: x=yield # time.sleep(1) #發現什麼？只是進行了切換，可是並無節省I/O時間 print('處理了數據：',x) def producer(): '''任務2:生產數據''' g=consumer() next(g) #找到了consumer函數的yield位置 for i in range(3): # for i in range(10000000): g.send(i) #給yield傳值，而後再循環給下一個yield傳值，而且多了切換的程序，比直接串行執行還多了一些步驟，致使執行效率反而更低了。 print('發送了數據：',i) start=time.time() #基於yield保存狀態,實現兩個任務直接來回切換,即併發的效果 #PS:若是每一個任務中都加上打印,那麼明顯地看到兩個任務的打印是你一次我一次,即併發執行的. producer() #我在當前線程中只執行了這個函數，可是經過這個函數裏面的send切換了另一個任務 stop=time.time() # 串行執行的方式 # res=producer() # consumer(res) # stop=time.time() print(stop-start) 

Python

Copy

二、協程就是告訴Cpython解釋器，你不是nb嗎，不是搞了個GIL鎖嗎，

那好，我就本身搞成一個線程讓你去執行，省去你切換線程的時間，我本身切換比你切換要快不少，避免了不少的開銷，
對於單線程下，咱們不可避免程序中出現io操做，但若是咱們能在本身的程序中（即用戶程序級別，而非操做系統級別）
控制單線程下的多個任務能在一個任務遇到io阻塞時就切換到另一個任務去計算，
這樣就保證了該線程可以最大限度地處於就緒態，即隨時均可以被cpu執行的狀態，
至關於咱們在用戶程序級別將本身的io操做最大限度地隱藏起來，從而能夠迷惑操做系統，
讓其看到：該線程好像是一直在計算，io比較少，從而更多的將cpu的執行權限分配給咱們的線程。

　　協程的本質就是在單線程下，由用戶本身控制一個任務遇到io阻塞了就切換另一個任務去執行，
以此來提高效率。爲了實現它，咱們須要找尋一種能夠同時知足如下條件的解決方案：

#1. 能夠控制多個任務之間的切換，切換以前將任務的狀態保存下來，以便從新運行時，能夠基於暫停的位置繼續執行。

#2. 做爲1的補充：能夠檢測io操做，在遇到io操做的狀況下才發生切換

三、協程介紹

協程：是單線程下的併發，又稱微線程，纖程。英文名Coroutine。
一句話說明什麼是線程：協程是一種用戶態的輕量級線程，即協程是由用戶程序本身控制調度的。

對比操做系統控制線程的切換，用戶在單線程內控制協程的切換
協程在操做系統上是沒有這個概念的，是程序員們本身叫的

1.須要強調的是：

1. python的線程屬於內核級別的，即由操做系統控制調度
（如單線程遇到io或執行時間過長就會被迫交出cpu執行權限，切換其餘線程運行）

2. 單線程內開啓協程，一旦遇到io，就會從應用程序級別（而非操做系統）控制切換，
以此來提高效率（！！！非io操做的切換與效率無關）

2.優勢以下：

1. 協程的切換開銷更小，屬於程序級別的切換，操做系統徹底感知不到，於是更加輕量級
2. 單線程內就能夠實現併發的效果，最大限度地利用cpu

3.缺點以下：

1. 協程的本質是單線程下，沒法利用多核，
能夠是一個程序開啓多個進程，每一個進程內開啓多個線程，每一個線程內開啓協程
2. 協程指的是單個線程，於是一旦協程出現阻塞，將會阻塞整個線程

4.總結協程特色：

一、必須在只有一個單線程裏實現併發
二、修改共享數據不需加鎖
三、用戶程序裏本身保存多個控制流的上下文棧
四、附加：一個協程遇到IO操做自動切換到其它協程
（如何實現檢測IO，yield、greenlet都沒法實現，就用到了gevent模塊（select機制））

四、Greenlet

若是咱們在單個線程內有20個任務，要想實如今多個任務之間切換，
使用yield生成器的方式過於麻煩（須要先獲得初始化一次的生成器，而後再調用send。。。很是麻煩），
而使用greenlet模塊能夠很是簡單地實現這20個任務直接的切換

#安裝
pip3 install greenlet

1.生成器版的任務切換

import time def func(): for i in range(10): print("來這裏") yield # 經過yield記錄上一次執行的位置 time.sleep(1) print("%s號完成了" % i) def func1(): g = func() next(g) for i in range(10): time.sleep(1) print("%s號在忙" % i) next(g) # func1() 

Python

Copy

2.效率對比

單純的切換（在沒有io的狀況下或者沒有重複開闢內存空間的操做），反而會下降程序的執行速度

效率對比

#順序執行 import time def f1(): res=1 for i in range(100000000): res+=i def f2(): res=1 for i in range(100000000): res*=i start=time.time() f1() f2() stop=time.time() print('run time is %s' %(stop-start)) #10.985628366470337 #切換 from greenlet import greenlet import time def f1(): res=1 for i in range(100000000): res+=i g2.switch() def f2(): res=1 for i in range(100000000): res*=i g1.switch() start=time.time() g1=greenlet(f1) g2=greenlet(f2) g1.switch() stop=time.time() print('run time is %s' %(stop-start)) # 52.763017892837524 

Python

Copy

3.greenlet只是提供了一種便捷的切換方式，並無提高效率

greenlet只是提供了一種比generator更加便捷的切換方式，
當切到一個任務執行時若是遇到io，那就原地阻塞，仍然是沒有解決遇到IO自動切換來提高效率的問題。

上面這個圖，是協程真正的意義，雖然沒有規避固有的I/O時間，可是咱們使用這個時間來作別的事情了， 通常在工做中咱們都是進程+線程+協程的方式來實現併發，以達到最好的併發效果， 若是是4核的cpu，通常起5個進程，每一個進程中20個線程（5倍cpu數量），每一個線程能夠起500個協程， 大規模爬取頁面的時候，等待網絡延遲的時間的時候，咱們就能夠用協程去實現併發。 併發數量 = 5 * 20 * 500 = 50000個併發，這是通常一個4cpu的機器最大的併發數。 nginx在負載均衡的時候最大承載量就是5w個 單線程裏的這20個任務的代碼一般會既有計算操做又有阻塞操做，咱們徹底能夠在執行任務1時遇到阻塞， 就利用阻塞的時間去執行任務2。。。。如此，才能提升效率，這就用到了Gevent模塊。 

Python

Copy

五、Gevent介紹

#安裝
pip3 install gevent

　　Gevent 是一個第三方庫，能夠輕鬆經過gevent實現併發同步或異步編程，
在gevent中用到的主要模式是Greenlet,
它是以C擴展模塊形式接入Python的輕量級協程。
Greenlet所有運行在主程序操做系統進程的內部，但它們被協做式地調度。

一、用法

g1=gevent.spawn(func,1,2,3,x=4,y=5)建立一個協程對象g1，spawn括號內第一個參數是函數名， 如eat，後面能夠有多個參數，能夠是位置實參或關鍵字實參，都是傳給函數eat的，spawn是異步提交任務 g2=gevent.spawn(func2) g1.join() #等待g1結束，上面只是建立協程對象，這個join纔是去執行;若是不寫會由於主代碼太快，而沒有被執行 g2.join() #等待g2結束 有人測試的時候會發現，不寫第二個join也能執行g2，是的，協程幫你切換執行了， 可是你會發現，若是g2裏面的任務執行的時間長，可是不寫join的話，就不會執行完等到g2剩下的任務了 #或者上述兩步合做一步：gevent.joinall([g1,g2]) g1.value#拿到func1的返回值 

Python

Copy

二、gevent的簡單使用

from gevent import monkey; monkey.patch_all() import time import gevent def func1(n): print("xxxxx", n) gevent.sleep(2) # time.sleep(2) # 沒有使用monkey,模塊不能識別其餘io切換，monkey模塊的做用是標記出來 print("eeeee", n) def func2(m): print("11111", m) gevent.sleep(2) # time.sleep(2) print("222222", m) start_time = time.time() g1 = gevent.spawn(func1, "李白") g2 = gevent.spawn(func2, "疏影") # g1.join() # g2.join() gevent.joinall([g1,g2]) # 實現了單線程的併發（線程裏面的協程），節省了時間 end_time = time.time() print("運行時間", end_time - start_time) print("主任務結束") ```` #### 上例gevent.sleep(2)模擬的是gevent能夠識別的io阻塞, 而time.sleep(2)或其餘的阻塞,gevent是不能直接識別的須要用下面一行代碼,打補丁,就能夠識別了 from gevent import monkey;monkey.patch_all()必須放到被打補丁者的前面，如time，socket模塊以前 能夠直接理解爲：必須放在最文件開頭，才能標記後面的io阻塞 或者咱們乾脆記憶成：要用gevent，須要將from gevent import monkey;monkey.patch_all()放到文件的開頭 ```python 咱們能夠用threading.current_thread().getName()來查看每一個g1和g2， 查看的結果爲DummyThread-n，即假線程，虛擬線程，其實都在一個線程裏面 進程線程的任務切換是由操做系統自行切換的，你本身不能控制 協程是經過本身的程序（代碼）來進行切換的，本身可以控制， 只有遇到協程模塊可以識別的IO操做的時候，程序纔會進行任務切換， 實現併發效果，若是全部程序都沒有IO操做，那麼就基本屬於串行執行了。 

Python

Copy

三、 Gevent之同步與異步

協程：同步異步對比

from gevent import spawn,joinall,monkey;monkey.patch_all() import time def task(pid): """ Some non-deterministic task """ time.sleep(0.5) print('Task %s done' % pid) def synchronous(): for i in range(10): task(i) def asynchronous(): g_l=[spawn(task,i) for i in range(10)] joinall(g_l) if __name__ == '__main__': print('Synchronous:') synchronous() print('Asynchronous:') asynchronous() #上面程序的重要部分是將task函數封裝到Greenlet內部線程的gevent.spawn。 初始化的greenlet列表存放在數組threads中，此數組被傳給gevent.joinall 函數， 後者阻塞當前流程，並執行全部給定的greenlet。執行流程只會在 全部greenlet執行完後纔會繼續向下走。 

Python

Copy

四、Gevent之應用舉例一

協程應用：爬蟲

from gevent import monkey;monkey.patch_all() import gevent import requests import time def get_page(url): print('GET: %s' %url) response=requests.get(url) if response.status_code == 200: print('%d bytes received from %s' %(len(response.text),url)) start_time=time.time() gevent.joinall([ gevent.spawn(get_page,'https://www.python.org/'), gevent.spawn(get_page,'https://www.yahoo.com/'), gevent.spawn(get_page,'https://github.com/'), ]) stop_time=time.time() print('run time is %s' %(stop_time-start_time)) 

Python

Copy

將上面的程序最後加上一段串行的代碼看看效率：
若是你的程序不須要過高的效率，那就不用什麼併發啊協程啊之類的東西。

print('--------------------------------') s = time.time() requests.get('https://www.python.org/') requests.get('https://www.yahoo.com/') requests.get('https://github.com/') t = time.time() print('串行時間>>',t-s) 

Python

Copy

五、Gevent之應用舉例二

經過gevent實現單線程下的socket併發（from gevent import monkey;monkey.patch_all()
必定要放到導入socket模塊以前，不然gevent沒法識別socket的阻塞）

一個網絡請求裏面通過多個時間延遲time

服務端

from gevent import monkey;monkey.patch_all() from socket import * import gevent #若是不想用money.patch_all()打補丁,能夠用gevent自帶的socket # from gevent import socket # s=socket.socket() def server(server_ip,port): s=socket(AF_INET,SOCK_STREAM) s.setsockopt(SOL_SOCKET,SO_REUSEADDR,1) s.bind((server_ip,port)) s.listen(5) while True: conn,addr=s.accept() gevent.spawn(talk,conn,addr) def talk(conn,addr): try: while True: res=conn.recv(1024) print('client %s:%s msg: %s' %(addr[0],addr[1],res)) conn.send(res.upper()) except Exception as e: print(e) finally: conn.close() if __name__ == '__main__': server('127.0.0.1',8080) 

Python

Copy

客戶端

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

Python

Copy

多線程併發多個客戶端，去請求上面的服務端是沒問題的

from threading import Thread from socket import * import threading def client(server_ip,port): c=socket(AF_INET,SOCK_STREAM) #套接字對象必定要加到函數內，即局部名稱空間內，放在函數外則被全部線程共享，則你們公用一個套接字對象，那麼客戶端端口永遠同樣了 c.connect((server_ip,port)) count=0 while True: c.send(('%s say hello %s' %(threading.current_thread().getName(),count)).encode('utf-8')) msg=c.recv(1024) print(msg.decode('utf-8')) count+=1 if __name__ == '__main__': for i in range(500): t=Thread(target=client,args=('127.0.0.1',8080)) t.start() 多線程併發多個客戶端，去請求上面的服務端是沒問題的 

Python

Copy

2、IO多路複用

同步（synchronous） IO和異步（asynchronous） IO，阻塞（blocking） IO和非阻塞（non-blocking）IO分別是什麼，
到底有什麼區別？這個問題其實不一樣的人給出的答案均可能不一樣，
好比wiki，就認爲asynchronous IO和non-blocking IO是一個東西。
這實際上是由於不一樣的人的知識背景不一樣，而且在討論這個問題的時候上下文(context)也不相同。
因此，爲了更好的回答這個問題，我先限定一下本文的上下文。

　　本文討論的背景是Linux環境下的network IO。
本文最重要的參考文獻是Richard Stevens的「UNIX® Network Programming Volume 1, Third Edition: The Sockets Networking 」，
6.2節「I/O Models 」，Stevens在這節中詳細說明了各類IO的特色和區別，
若是英文夠好的話，推薦直接閱讀。
Stevens的文風是有名的深刻淺出，因此不用擔憂看不懂。本文中的流程圖也是截取自參考文獻。

一、Stevens在文章中一共比較了五種IO Model：

* blocking IO          阻塞IO

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

　　再說一下IO發生時涉及的對象和步驟。對於一個network IO (這裏咱們以read、recv舉例)，它會涉及到兩個系統對象，一個是調用這個IO的process (or thread)，另外一個就是系統內核(kernel)。當一個read/recv讀數據的操做發生時，該操做會經歷兩個階段：

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

　　記住這兩點很重要，由於這些IO模型的區別就是在兩個階段上各有不一樣的狀況。

1.補充：

#一、輸入操做：read、readv、recv、recvfrom、recvmsg共5個函數， 若是會阻塞狀態，則會經理wait data和copy data兩個階段，若是設置爲非阻塞則在wait 不到data時拋出異常 #二、輸出操做：write、writev、send、sendto、sendmsg共5個函數， 在發送緩衝區滿了會阻塞在原地，若是設置爲非阻塞，則會拋出異常 #三、接收外來連接：accept，與輸入操做相似 #四、發起外出連接：connect，與輸出操做相似 

Python

Copy

二、阻塞IO(blocking IO)

就是咱們日常寫的input,的代碼，這樣的阻塞
在linux中，默認狀況下全部的socket都是blocking，
一個典型的讀操做流程大概是這樣：（recvfrom和tcp裏面的recv在這些IO模型裏面是同樣的）

上面的圖形分析：兩個階段的阻塞
因此，blocking IO的特色就是在IO執行的兩個階段（等待數據和拷貝數據兩個階段）都被block了。

　　這裏咱們回顧一下同步/異步/阻塞/非阻塞：

　　　　同步：提交一個任務以後要等待這個任務執行完畢

　　　　異步：只管提交任務，不等待這個任務執行完畢就能夠去作其餘的事情

　　　　阻塞：recv、recvfrom、accept，線程階段 運行狀態–>阻塞狀態–>就緒

　　　　非阻塞：沒有阻塞狀態

在一個線程的IO模型中，咱們recv的地方阻塞，咱們就開啓多線程，
可是無論你開啓多少個線程，這個recv的時間是否是沒有被規避掉，
無論是多線程仍是多進程都沒有規避掉這個IO時間。

三、非阻塞IO

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

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

雖然咱們上面的代碼經過設置非阻塞，規避了IO操做，可是非阻塞IO模型毫不被推薦。

非阻塞IO模型服務端

import time import socket server = socket.socket() ip_port = ("127.0.0.1", 8001) server.bind(ip_port) server.listen() server.setblocking(False) # 設置不阻塞，把會阻塞的地方變得不阻塞，可是會報錯，要用try捕獲異常 conn_list = [] # 存放服務端對象，已經鏈接的管道對象 while 1: while 1: # 不停的輪詢內核， try: conn,addr = server.accept() # 使用了setblocking後，這裏不會阻塞，會有協程一直輪詢內核， conn_list.append(conn) # 將 break except BlockingIOError: time.sleep(0.2) print("尚未接收到鏈接請求") print("你好") time.sleep(5) for sock in conn_list: print(sock) # 管道對象 while 1: try: from_cilent_msg = conn.recv(1024).decode("utf-8") print(from_cilent_msg) conn.send(b"hello") break except BlockingIOError: time.sleep(0.2) # while循環很耗內存，用sleep緩衝一下 print("沒有接收到任何消息") 

Python

Copy

非阻塞IO模型客戶端

import socket cilent = socket.socket() ip_port = ("127.0.0.1", 8001) cilent.connect(ip_port) cilent_msg = input("請輸入消息>>>>") cilent.send(cilent_msg.encode("utf-8")) from_server_msg = cilent.recv(1024).decode("utf-8") print(from_server_msg) 

Python

Copy

非阻塞IO示例詳細版

# 服務端 import socket import time server=socket.socket() server.setsockopt(socket.SOL_SOCKET,socket.SO_REUSEADDR,1) server.bind(('127.0.0.1',8083)) server.listen(5) server.setblocking(False) #設置不阻塞 r_list=[] #用來存儲全部來請求server端的conn鏈接 w_list={} #用來存儲全部已經有了請求數據的conn的請求數據 while 1: try: conn,addr=server.accept() #不阻塞，會報錯 r_list.append(conn) #爲了將鏈接保存起來，否則下次循環的時候，上一次的鏈接就沒有了 except BlockingIOError: # 強調強調強調：！！！非阻塞IO的精髓在於徹底沒有阻塞！！！ # time.sleep(0.5) # 打開該行註釋純屬爲了方便查看效果 print('在作其餘的事情') print('rlist: ',len(r_list)) print('wlist: ',len(w_list)) # 遍歷讀列表，依次取出套接字讀取內容 del_rlist=[] #用來存儲刪除的conn鏈接 for conn in r_list: try: data=conn.recv(1024) #不阻塞，會報錯 if not data: #當一個客戶端暴力關閉的時候，會一直接收b''，別忘了判斷一下數據 conn.close() del_rlist.append(conn) continue w_list[conn]=data.upper() except BlockingIOError: # 沒有收成功，則繼續檢索下一個套接字的接收 continue except ConnectionResetError: # 當前套接字出異常，則關閉，而後加入刪除列表，等待被清除 conn.close() del_rlist.append(conn) # 遍歷寫列表，依次取出套接字發送內容 del_wlist=[] for conn,data in w_list.items(): try: conn.send(data) del_wlist.append(conn) except BlockingIOError: continue # 清理無用的套接字,無需再監聽它們的IO操做 for conn in del_rlist: r_list.remove(conn) #del_rlist.clear() #清空列表中保存的已經刪除的內容 for conn in del_wlist: w_list.pop(conn) #del_wlist.clear() #客戶端 import socket import os import time import threading client=socket.socket() client.connect(('127.0.0.1',8083)) while 1: res=('%s hello' %os.getpid()).encode('utf-8') client.send(res) data=client.recv(1024) print(data.decode('utf-8')) ##多線程的客戶端請求版本 # def func(): # sk = socket.socket() # sk.connect(('127.0.0.1',9000)) # sk.send(b'hello') # time.sleep(1) # print(sk.recv(1024)) # sk.close() # # for i in range(20): # threading.Thread(target=func).start() 

Python

Copy

1.非阻塞IO模型毫不被推薦

雖然咱們上面的代碼經過設置非阻塞，規避了IO操做，可是非阻塞IO模型毫不被推薦。

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

2.可是也難掩其缺點：

1. 循環調用recv()將大幅度推高CPU佔用率；
這也是咱們在代碼中留一句time.sleep(2)的緣由,不然在低配主機下極容易出現卡機狀況

2. 任務完成的響應延遲增大了，由於每過一段時間纔去輪詢一次read操做，
而任務可能在兩次輪詢之間的任意時間完成。這會致使總體數據吞吐量的下降。

此外，在這個方案中recv()更多的是起到檢測「操做是否完成」的做用，
實際操做系統提供了更爲高效的檢測「操做是否完成「做用的接口，
例如select()多路複用模式，能夠一次檢測多個鏈接是否活躍。

四、多路複用IO(IO multiplexing)（重點）

先看解釋圖，裏面的select就像個代理。

IO multiplexing這個詞可能有點陌生，可是若是我說select/epoll，大概就都能明白了。
有些地方也稱這種IO方式爲事件驅動IO(event driven IO)。
咱們都知道，select/epoll的好處就在於單個process就能夠同時處理多個網絡鏈接的IO。
它的基本原理就是select/epoll這個function會不斷的輪詢所負責的全部socket，
當某個socket有數據到達了，就通知用戶進程。它的流程如圖：

1.強調：

1. 若是處理的鏈接數不是很高的話，使用select/epoll的web server不必定比使用multi-threading + blocking IO的web server性能更好，
可能延遲還更大。select/epoll的優點並非對於單個鏈接能處理得更快，而是在於能處理更多的鏈接。

2. 在多路複用模型中，對於每個socket，通常都設置成爲non-blocking，可是，
如上圖所示，整個用戶的process實際上是一直被block的。
只不過process是被select這個函數block，而不是被socket IO給block。

2.python中的select模塊：

import select fd_r_list, fd_w_list, fd_e_list = select.select(rlist, wlist, xlist, [timeout]) 參數： 可接受四個參數（前三個必須） rlist: wait until ready for reading #等待讀的對象，你須要監聽的須要獲取數據的對象列表 wlist: wait until ready for writing #等待寫的對象，你須要寫一些內容的時候，input等等， 也就是說我會循環他看看是否有須要發送的消息，若是有我取出這個對象的消息併發送出去，通常用不到，這裏咱們也給一個[]。 xlist: wait for an 「exceptional condition」 #等待異常的對象，一些額外的狀況，通常用不到，可是必須傳，那麼咱們就給他一個[]。 timeout: 超時時間 當超時時間 ＝ n(正整數)時，那麼若是監聽的句柄均無任何變化，則select會阻塞n秒，以後返回三個空列表，若是監聽的句柄有變化，則直接執行。 返回值：三個列表與上面的三個參數列表是對應的 　　select方法用來監視文件描述符(當文件描述符條件不知足時，select會阻塞)，當某個文件描述符狀態改變後，會返回三個列表 一、當參數1 序列中的fd知足「可讀」條件時，則獲取發生變化的fd並添加到fd_r_list中 二、當參數2 序列中含有fd時，則將該序列中全部的fd添加到 fd_w_list中 三、當參數3 序列中的fd發生錯誤時，則將該發生錯誤的fd添加到 fd_e_list中 四、當超時時間爲空，則select會一直阻塞，直到監聽的句柄發生變化 

Python

Copy

三、結論:

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

fe1：io多路複用服務端、客戶端示例

io多路複用服務端

import select import socket server = socket.socket() server.bind(('127.0.0.1',8001)) rlist = [server,] # 先將server對象放入列表，而後再將每一個通道對象放入 server.listen() while 1: print('11111') rl,wl,el = select.select(rlist,[],[]) # 每次會清空rl列表，從新監測 print(222222) print('server對象>>>',server) print(rl) for sock in rl: # 是server對象就添加通道對象，是通道，就行對應的收發消息 if sock == server: conn,addr = sock.accept() rlist.append(conn) else: from_client_msg = sock.recv(1024) print(from_client_msg.decode('utf-8')) # conn,addr = server.accept() # from_client_msg = conn.recv(1024) # print(from_client_msg.decode('utf-8')) 

Python

Copy

io多路複用客戶端

import socket client = socket.socket() client.connect(('127.0.0.1',8001)) to_server_msg = input('發給服務端的消息:') client.send(to_server_msg.encode('utf-8')) # from_server_msg = client.recv(1024) # print(from_server_msg.decode('utf-8')) 

Python

Copy

select網絡IO模型的示例代碼詳細解釋版

#服務端 from socket import * import select server = socket(AF_INET, SOCK_STREAM) server.bind(('127.0.0.1',8093)) server.listen(5) # 設置爲非阻塞 server.setblocking(False) # 初始化將服務端socket對象加入監聽列表，後面還要動態添加一些conn鏈接對象，當accept的時候sk就有感應，當recv的時候conn就有動靜 rlist=[server,] rdata = {} #存放客戶端發送過來的消息 wlist=[] #等待寫對象 wdata={} #存放要返回給客戶端的消息 print('預備！監聽！！！') count = 0 #寫着計數用的，爲了看實驗效果用的，沒用 while True: # 開始 select 監聽,對rlist中的服務端server進行監聽，select函數阻塞進程，直到rlist中的套接字被觸發（在此例中，套接字接收到客戶端發來的握手信號，從而變得可讀，知足select函數的「可讀」條件），被觸發的（有動靜的）套接字（服務器套接字）返回給了rl這個返回值裏面； rl,wl,xl=select.select(rlist,wlist,[],0.5) print('%s 次數>>'%(count),wl) count = count + 1 # 對rl進行循環判斷是否有客戶端鏈接進來,當有客戶端鏈接進來時select將觸發 for sock in rl: # 判斷當前觸發的是否是socket對象, 當觸發的對象是socket對象時,說明有新客戶端accept鏈接進來了 if sock == server: # 接收客戶端的鏈接, 獲取客戶端對象和客戶端地址信息 conn,addr=sock.accept() #把新的客戶端鏈接加入到監聽列表中，當客戶端的鏈接有接收消息的時候，select將被觸發，會知道這個鏈接有動靜，有消息，那麼返回給rl這個返回值列表裏面。 rlist.append(conn) else: # 因爲客戶端鏈接進來時socket接收客戶端鏈接請求，將客戶端鏈接加入到了監聽列表中(rlist)，客戶端發送消息的時候這個鏈接將觸發 # 因此判斷是不是客戶端鏈接對象觸發 try: data=sock.recv(1024) #沒有數據的時候，咱們將這個鏈接關閉掉，並從監聽列表中移除 if not data: sock.close() rlist.remove(sock) continue print("received {0} from client {1}".format(data.decode(), sock)) #將接受到的客戶端的消息保存下來 rdata[sock] = data.decode() #將客戶端鏈接對象和這個對象接收到的消息加工成返回消息，並添加到wdata這個字典裏面 wdata[sock]=data.upper() #須要給這個客戶端回覆消息的時候，咱們將這個鏈接添加到wlist寫監聽列表中 wlist.append(sock) #若是這個鏈接出錯了，客戶端暴力斷開了（注意，我尚未接收他的消息，或者接收他的消息的過程當中出錯了） except Exception: #關閉這個鏈接 sock.close() #在監聽列表中將他移除，由於無論什麼緣由，它畢竟是斷開了，不必再監聽它了 rlist.remove(sock) # 若是如今沒有客戶端請求鏈接,也沒有客戶端發送消息時，開始對發送消息列表進行處理，是否須要發送消息 for sock in wl: sock.send(wdata[sock]) wlist.remove(sock) wdata.pop(sock) # #將一次select監聽列表中有接收數據的conn對象所接收到的消息打印一下 # for k,v in rdata.items(): # print(k,'發來的消息是：',v) # #清空接收到的消息 # rdata.clear() --------------------------------------- #客戶端 from socket import * client=socket(AF_INET,SOCK_STREAM) client.connect(('127.0.0.1',8093)) 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() 

Python

Copy

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

用戶進程建立socket對象，拷貝監聽的fd到內核空間，每個fd會對應一張系統文件表，
內核空間的fd響應到數據後，就會發送信號給用戶進程數據已到；

用戶進程再發送系統調用，好比（accept）將內核空間的數據copy到用戶空間，同時做爲接受數據端內核空間的數據清除，
這樣從新監聽時fd再有新的數據又能夠響應到了（發送端由於基於TCP協議因此須要收到應答後纔會清除）。

五、該模型的優勢：

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

六、該模型的缺點：

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

其次，該模型將事件探測和事件響應夾雜在一塊兒，一旦事件響應的執行體龐大，則對整個模型是災難性的。

七、IO多路複用的機制：

Select  將全部須要監聽的對象,放到一個列表裏面,將這個列表交給select來監聽,
凡有動靜的對象,直接給你返回到一個列表中,而後咱們循環這個列表,根據列表裏面的有動靜的對象,來進行對應的操做.  window 和 linux

Poll  : 監聽的對象數量沒有上限,而select默認是1024個,Linux
    Select和poll內部在監聽的全部對象的時候,是循環遍歷的操做,挨個問一遍有沒有數據

Epoll : linux,每一個被監聽的對象,都是經過回調機制來搞的,
也就是,監聽的列表對象裏面的這個對象,凡有動靜,自行通知epoll,而後epoll把這些有動靜的對象返回.

五、異步IO(Asynchronous I/O)

最NB的,效率最高,由於將網絡請求裏面的兩個io時間都省了