十四、并發(fā)編程之I/O模型

http://www.cnblogs.com/linhaifeng/articles/7454717.html

1、模型介紹

1.1 IO種類

(1)* blocking IO 阻塞IO

(2)* nonblocking IO 非阻塞IO

(3)* IO multiplexing 多路復用IO

(4)* signal driven IO 信號驅(qū)動IO

(5)* asynchronous IO 異步IO

1.2 IO發(fā)生時的對象和步驟

對于一個network IO (這里我們以read舉例)，它會涉及到兩個系統(tǒng)對象，一個是調(diào)用這個IO的process (or thread)，另一個就是系統(tǒng)內(nèi)核(kernel)。

1.3 IO經(jīng)歷的階段

(1)等待數(shù)據(jù)準備

(2)將數(shù)據(jù)從內(nèi)核(操作系統(tǒng)內(nèi)存)拷貝到進程(應用程序內(nèi)存)中

2、阻塞IO

在linux中，默認情況下所有的socket都是blocking，一個典型的讀操作流程大概是這樣：

blocking IO的特點就是在IO執(zhí)行的兩個階段(等待數(shù)據(jù)和拷貝數(shù)據(jù)兩個階段)都被block了。

3、非阻塞IO

3.1 流程

Linux下，可以通過設置socket使其變?yōu)閚on-blocking。當對一個non-blocking socket執(zhí)行讀操作時，流程是這個樣子：

在非阻塞式IO中，用戶進程其實是需要不斷的主動詢問kernel數(shù)據(jù)準備好了沒有。 ?但是非阻塞IO模型絕不被推薦。

3.2 實例

#服務端fromsocket import*

importtime

s=socket(AF_INET,SOCK_STREAM)

s.bind(('127.0.0.1',8080))

s.listen(5)

s.setblocking(False) #設置socket的接口為非阻塞conn_l=[]

del_l=[]

while True:

try:

conn,addr=s.accept()

conn_l.append(conn)

exceptBlockingIOError:

print(conn_l)

forconn inconn_l:

try:

data=conn.recv(1024)

if notdata:

del_l.append(conn)

continueconn.send(data.upper())

exceptBlockingIOError:

passexceptConnectionResetError:

del_l.append(conn)

forconn indel_l:

conn_l.remove(conn)

conn.close()

del_l=[]

#客戶端fromsocket import*

c=socket(AF_INET,SOCK_STREAM)

c.connect(('127.0.0.1',8080))

while True:

msg=input('>>: ')

if notmsg:continuec.send(msg.encode('utf-8'))

data=c.recv(1024)

print(data.decode('utf-8'))

3.3 優(yōu)缺點

優(yōu)點：能夠在等待任務完成的時間里干其他活了

缺點：

#1. 循環(huán)調(diào)用recv()將大幅度推高CPU占用率；這也是我們在代碼中留一句time.sleep(2)的原因,否則在低配主機下極容易出現(xiàn)卡機情況

#2. 任務完成的響應延遲增大了，因為每過一段時間才去輪詢一次read操作，而任務可能在兩次輪詢之間的任意時間完成。這會導致整體數(shù)據(jù)吞吐量的降低。

4、多路復用IO

4.1 流程

有些地方也稱這種IO方式為事件驅(qū)動IO(event driven IO)。我們都知道，select/epoll的好處就在于單個process就可以同時處理多個網(wǎng)絡連接的IO。它的基本原理就是select/epoll這個function會不斷的輪詢所負責的所有socket，當某個socket有數(shù)據(jù)到達了，就通知用戶進程。它的流程如圖：

當用戶進程調(diào)用了select，那么整個進程會被block，而同時，kernel會“監(jiān)視”所有select負責的socket，當任何一個socket中的數(shù)據(jù)準備好了，select就會返回。這個時候用戶進程再調(diào)用read操作，將數(shù)據(jù)從kernel拷貝到用戶進程。

強調(diào)：

1. 如果處理的連接數(shù)不是很高的話，使用select/epoll的web server不一定比使用multi-threading + blocking IO的web server性能更好，可能延遲還更大。select/epoll的優(yōu)勢并不是對于單個連接能處理得更快，而是在于能處理更多的連接。

2. 在多路復用模型中，對于每一個socket，一般都設置成為non-blocking，但是，如上圖所示，整個用戶的process其實是一直被block的。只不過process是被select這個函數(shù)block，而不是被socket IO給block。

結(jié)論: select的優(yōu)勢在于可以處理多個連接，不適用于單個連接

4.2 實例

#服務端fromsocket import*

importselect

s=socket(AF_INET,SOCK_STREAM)

s.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)

s.bind(('127.0.0.1',8081))

s.listen(5)

s.setblocking(False) #設置socket的接口為非阻塞read_l=[s,]

while True:

r_l,w_l,x_l=select.select(read_l,[],[])

print(r_l)

forready_obj inr_l:

ifready_obj == s:

conn,addr=ready_obj.accept() #此時的ready_obj等于sread_l.append(conn)

else:

try:

data=ready_obj.recv(1024) #此時的ready_obj等于connif notdata:

ready_obj.close()

read_l.remove(ready_obj)

continueready_obj.send(data.upper())

exceptConnectionResetError:

ready_obj.close()

read_l.remove(ready_obj)

#客戶端fromsocket import*

c=socket(AF_INET,SOCK_STREAM)

c.connect(('127.0.0.1',8081))

while True:

msg=input('>>: ')

if notmsg:continuec.send(msg.encode('utf-8'))

data=c.recv(1024)

print(data.decode('utf-8'))

4.3 優(yōu)缺點

優(yōu)點：

#相比其他模型，使用select()的事件驅(qū)動模型只用單線程(進程)執(zhí)行，占用資源少，不消耗太多CPU，同時能夠為多客戶端提供服務。如果試圖建立一個簡單的事件驅(qū)動的服務器程序，這個模型有一定的參考價值。

缺點：

#首先select()接口并不是實現(xiàn)“事件驅(qū)動”的最好選擇。因為當需要探測的句柄值較大時，select()接口本身需要消耗大量時間去輪詢各個句柄。很多操作系統(tǒng)提供了更為高效的接口，如linux提供了epoll，BSD提供了kqueue，Solaris提供了/dev/poll，…。如果需要實現(xiàn)更高效的服務器程序，類似epoll這樣的接口更被推薦。遺憾的是不同的操作系統(tǒng)特供的epoll接口有很大差異，所以使用類似于epoll的接口實現(xiàn)具有較好跨平臺能力的服務器會比較困難。

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

5、異步IO

Linux下的asynchronous IO其實用得不多，從內(nèi)核2.6版本才開始引入。先看一下它的流程：

用戶進程發(fā)起read操作之后，立刻就可以開始去做其它的事。而另一方面，從kernel的角度，當它受到一個asynchronous read之后，首先它會立刻返回，所以不會對用戶進程產(chǎn)生任何block。然后，kernel會等待數(shù)據(jù)準備完成，然后將數(shù)據(jù)拷貝到用戶內(nèi)存，當這一切都完成之后，kernel會給用戶進程發(fā)送一個signal，告訴它read操作完成了。

6、selectors模塊

select,poll,epoll這三種IO多路復用模型在不同的平臺有著不同的支持，而epoll在windows下就不支持，好在我們有selectors模塊，幫我們默認選擇當前平臺下最合適的

#服務端fromsocket import*

importselectors

sel=selectors.DefaultSelector()

defaccept(server_fileobj,mask):

conn,addr=server_fileobj.accept()

sel.register(conn,selectors.EVENT_READ,read)

defread(conn,mask):

try:

data=conn.recv(1024)

if notdata:

print('closing',conn)

sel.unregister(conn)

conn.close()

returnconn.send(data.upper()+b'_SB')

exceptException:

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) #相當于網(wǎng)select的讀列表里append了一個文件句柄server_fileobj,并且綁定了一個回調(diào)函數(shù)acceptwhile True:

events=sel.select() #檢測所有的fileobj，是否有完成wait data的forsel_obj,mask inevents:

callback=sel_obj.data #callback=accpetcallback(sel_obj.fileobj,mask) #accpet(server_fileobj,1)#客戶端fromsocket import*

c=socket(AF_INET,SOCK_STREAM)

c.connect(('127.0.0.1',8088))

while True:

msg=input('>>: ')

if notmsg:continuec.send(msg.encode('utf-8'))

data=c.recv(1024)

print(data.decode('utf-8'))

7、socketsever模塊

服務端：

importsocketserver

classMyTCPhandler(socketserver.BaseRequestHandler):

defhandle(self):

# print(self)# print(self.request) #self.request=connwhile True:

try:

print(self.client_address)

print(self.server) #objdata=self.request.recv(1024)

if notdata:breakself.request.send(data.upper())

exceptConnectionResetError:

breakif__name__ == '__main__':

print('starting...')

obj=socketserver.ThreadingTCPServer(('127.0.0.1',8080),MyTCPhandler)

#bind_and_activate=True#socketserver.ForkingTCPServerobj.serve_forever() #連接循環(huán)'''t=Thread(target=MyTCPhandler().handle)t.start()'''# obj.server_address #('127.0.0.1',8080)# obj.RequestHandlerClass #MyTCPhandler# obj.socket #s對象

客戶端：

fromsocket import*

c=socket(AF_INET,SOCK_STREAM)

c.connect(('127.0.0.1',8080))

while True:

msg=input('>>: ').strip()

if notmsg:continuec.send(msg.encode('utf-8'))

data=c.recv(1024)

print(data.decode('utf-8'))

8、并發(fā)編程總結(jié)(重點)

1 生產(chǎn)者消費者模型

2 開啟進程、開啟線程

3 進程池、線程池(回調(diào)機制)

4 GIL全局解釋器鎖

總結(jié)

以上是生活随笔為你收集整理的python电路模型编程_14、python开发之路-并发编程之I/O模型的全部內(nèi)容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網(wǎng)站內(nèi)容還不錯，歡迎將生活随笔推薦給好友。