4.6 併發編程/IO模型

 併發編程/IO模型

背景概念

IO模型概念

IO模型分類

阻塞IO  (blocking IO)

特色： 

　　　　兩個階段（等待數據和拷貝數據兩個階段）都被block

設置

server.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)

解決方案： 

　　　　啓用多線程或者多進程，要阻塞只阻塞當前線程/進程，不會影響其餘進程/線程

不良影響：

　　　　當遇到過多得連接請求時會嚴重佔用資源，下降響應效率

修復不良影響：

　　　　啓用進程池/線程池 ，下降進程/線程數量

　仍舊未解決的不良影響：

　　　　池得數量很差規範，請求數量和池大小很差對應，且池存在數量上限

　　　　只是必定程度上限制了不良影響，沒法根本解決

　　總結 :

　　　　基於池得建立能夠解決小規模得服務請求帶來的壓力，對於大規模仍是無力迴天

非阻塞IO  (nonblocking IO)

　　特色： 

　　　　基於IO阻塞模型相似，再本應阻塞得地方不在阻塞，若是未收到想要數據會返回一個 error 給用戶告知無數據

　　　　發送 error 後會進行其餘得任務繼續操做背後會一直對 kernel 進行輪詢發送數據請求，

　　　　這期間若是數據到了就會從新正常操做，而在輪詢期間，無數據得進程會一直處於阻塞狀態

　　　　就結果而言。徹底得實現了併發，以及解決了IO阻塞帶來得效率低下的問題

設置

server.setblocking()    #默認是True  
server.setblocking(False)     #False的話就成非阻塞了，這只是對於socket套接字來講的 

不良影響：

　　　　1. 雖然說解決了單線程併發，可是大大的佔用了cpu

　　　　2.  任務完成的響應延遲增大，任務可能在兩次輪詢之間的任意時間完成。這會致使總體數據吞吐量的下降

　　總結 :

　　　　徹底不推薦

多路複用IO  (IO multiplexing)

　　特色： 

　　　　當用戶進程調用了select，那麼整個進程會被block

　　　　而同時，kernel會「監視」全部select負責的socket，當任何一個socket中的數據準備好了，select就會返回。

　　　　這個時候用戶進程再調用read操做，將數據從kernel拷貝到用戶進程。

　　　　這個圖和blocking IO的圖其實並無太大的不一樣，事實上還更差一些。

　　　　由於這裏須要使用兩個系統調用(select和recvfrom)，而blocking IO只調用了一個系統調用(recvfrom)。可是，用select的優點在於它能夠同時處理多個connection。

設置

使用 select 模塊 或者 eppol （只能用於 linux 中）
最優的選擇方案是用 selectors

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

View Code

不良影響：

　　　　1. 雖然說解決了單線程併發，可是大大的佔用了cpu

　　　　2.  任務完成的響應延遲增大，任務可能在兩次輪詢之間的任意時間完成。這會致使總體數據吞吐量的下降

　　總結 :

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

異步IO(asynchronous IO)

　　用戶進程發起read操做以後，馬上就能夠開始去作其它的事。

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

　　而後，kernel會等待數據準備完成，而後將數據拷貝到用戶內存，

　　當這一切都完成以後，kernel會給用戶進程發送一個signal，告訴它read操做完成了。

socketserver 模塊

 包含類

　　BaseServer是基類，它不能實例化使用，

　　TCPServer使用TCP協議通訊，

　　UDPServer使用UDP協議通訊，

　　UnixStreamServer和UnixDatagramServer使用Unix域套接字，只適用於UNIX平臺。

　　ForkingMixIn 多進程異步

　　ThreadingMixIn 多線程異步 

如何建立一個socketserver ：

　　1. 建立一個請求處理的類，繼承 BaseRequestHandlerclass ，要重寫父類裏 handle() 方法；

　　2. 你必須實例化 TCPServer，而且傳遞server IP和你上面建立的請求處理類，給這個TCPServer；

　　3. server.handle_requese()  只處理一個請求，server.server_forever() 處理多個一個請求，永遠執行

　　4. 關閉鏈接 server_close()

import socketserver

class MyTCPHandler(socketserver.BaseRequestHandler): #服務類，監聽綁定等等
  
 
    def handle(self):  #請求處理類，全部請求的交互都是在handle裏執行的
        # self.request is the TCP socket connected to the client
        self.data = self.request.recv(1024).strip()　　#每個請求都會實例化MyTCPHandler(socketserver.BaseRequestHandler):
        print("{} wrote:".format(self.client_address[0]))
        print(self.data)
        self.request.sendall(self.data.upper())　　#sendall是重複調用send.

if __name__ == "__main__":
    HOST, PORT = "localhost", 9999

    server = socketserver.TCPServer((HOST, PORT), MyTCPHandler)

    # Activate the server; this will keep running until you
    # interrupt the program with Ctrl-C
    server.serve_forever()
 
 server = socketserver.ThreadingTCPServer((HOST, PORT), MyTCPHandler)   #線程
# server  = socketserver.ForkingTCPServer((HOST, PORT), MyTCPHandler) #多進程 linux適用
# server = socketserver.TCPServer((HOST, PORT), MyTCPHandler) 單進程

ThreadingTCPServer

　　服務器內部會爲每一個client建立一個 「線程」，該線程用來和客戶端進行交互。

import SocketServer

class MyServer(SocketServer.BaseRequestHandler):

    def handle(self):
        pass

if __name__ == '__main__':
    server = SocketServer.ThreadingTCPServer(('127.0.0.1',8766), MyServer)
    server.serve_forever()

ForkingTCPServer

　　用法相似 ThreadingTCPServer 只是將線程換成了進程 

import SocketServer

class MyServer(SocketServer.BaseRequestHandler):

    def handle(self):
        pass

if __name__ == '__main__':
    server = SocketServer.ForkingTCPServer(('127.0.0.1',8009),MyServer)
    server.serve_forever()

 總結：

socketserver 完美的解決了以前的各類困擾。不管是併發問題仍是線程池大小以及IO問題都被解決。

　　所以socketserver成爲了最終解決方案