分析python并發網絡通信模型
循環接收客戶端請求,處理請求。同一時刻只能處理一個請求,處理完畢后再處理下一個。
優點:實現簡單,占用資源少 缺點:無法同時處理多個客戶端請求 適用情況:處理的任務可以很快完成,客戶端無需長期占用服務端程序。udp比tcp更適合循環。1.2、IO并發模型利用IO多路復用,異步IO等技術,同時處理多個客戶端IO請求。
優點 : 資源消耗少,能同時高效處理多個IO行為 缺點 : 只能處理并發產生的IO事件,無法處理cpu計算 適用情況:HTTP請求,網絡傳輸等都是IO行為。1.3、多進程/線程網絡并發模型每當一個客戶端連接服務器,就創建一個新的進程/線程為該客戶端服務,客戶端退出時再銷毀該進程/線程。
優點:能同時滿足多個客戶端長期占有服務端需求,可以處理各種請求。 缺點: 資源消耗較大 適用情況:客戶端同時連接量較少,需要處理行為較復雜情況。二、基于fork的多進程網絡并發模型1.創建監聽套接字
2.等待接收客戶端請求
3.客戶端連接創建新的進程處理客戶端請求
4.原進程繼續等待其他客戶端連接
5.如果客戶端退出,則銷毀對應的進程
from socket import *import osimport signal# 創建監聽套接字HOST = ’0.0.0.0’PORT = 8888ADDR = (HOST,PORT)# 客戶端服務函數def handle(c): while True: data = c.recv(1024) if not data: break print(data.decode()) c.send(b’OK’) c.close()s = socket() # tcp套接字s.setsockopt(SOL_SOCKET,SO_REUSEADDR,1) # 設置套接字端口重用s.bind(ADDR)s.listen(3)signal.signal(signal.SIGCHLD,signal.SIG_IGN) # 處理僵尸進程print('Listen the port %d...' % PORT)# 循環等待客戶端連接while True: try: c,addr = s.accept() except KeyboardInterrupt: os._exit(0) except Exception as e: print(e) continue # 創建子進程處理這個客戶端 pid = os.fork() if pid == 0: # 處理客戶端請求 s.close() handle(c) os._exit(0) # handle處理完客戶端請求子進程也退出 # 無論出錯或者父進程都要循環回去接受請求 # c對于父進程沒用 c.close()三、基于threading的多線程網絡并發
1.創建監聽套接字
2.循環接收客戶端連接請求
3.當有新的客戶端連接創建線程處理客戶端請求
4.主線程繼續等待其他客戶端連接
5.當客戶端退出,則對應分支線程退出
from socket import *from threading import Threadimport sys# 創建監聽套接字HOST = ’0.0.0.0’PORT = 8888ADDR = (HOST,PORT)# 處理客戶端請求def handle(c): while True: data = c.recv(1024) if not data: break print(data.decode()) c.send(b’OK’) c.close()s = socket() # tcp套接字s.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)s.bind(ADDR)s.listen(3)print('Listen the port %d...'%PORT)# 循環等待客戶端連接while True: try: c,addr = s.accept() except KeyboardInterrupt: sys.exit('服務器退出') except Exception as e: print(e) continue # 創建線程處理客戶端請求 t = Thread(target=handle, args=(c,)) t.setDaemon(True) # 父進程結束則所有進程終止 t.start()四、ftp 文件服務器4.1、項目功能
客戶端有簡單的頁面命令提示:功能包含：
查看服務器文件庫中的文件列表（普通文件） 可以下載其中的某個文件到本地 可以上傳客戶端文件到服務器文件庫服務器需求 ：
允許多個客戶端同時操作 每個客戶端可能回連續發送命令技術分析：
tcp套接字更適合文件傳輸 并發方案 ---》 fork 多進程并發 對文件的讀寫操作獲取 文件列表 ----》 os.listdir()粘包的處理
4.2、整體結構設計 服務器功能封裝在類中（上傳，下載，查看列表） 創建套接字，流程函數調用 main（） 客戶端負責發起請求，接受回復，展示 服務端負責接受請求，邏輯處理ftp server：
from socket import *from threading import Threadimport osimport time# 全局變量HOST = ’0.0.0.0’PORT = 8080ADDR = (HOST,PORT)FTP = '/home/tarena/FTP/' # 文件庫位置# 創建文件服務器服務端功能類class FTPServer(Thread): def __init__(self,connfd): self.connfd = connfd super().__init__() def do_list(self): # 獲取文件列表 files = os.listdir(FTP) if not files: self.connfd.send('文件庫為空'.encode()) return else: self.connfd.send(b’OK’) time.sleep(0.1) # 防止和后面發送內容粘包 # 拼接文件列表 files_ = '' for file in files: if file[0] != ’.’ and os.path.isfile(FTP+file):files_ += file + ’n’ self.connfd.send(files_.encode()) def do_get(self,filename): try: fd = open(FTP+filename,’rb’) except Exception: self.connfd.send('文件不存在'.encode()) return else: self.connfd.send(b’OK’) time.sleep(0.1) # 文件發送 while True: data = fd.read(1024) if not data:time.sleep(0.1)self.connfd.send(b’##’)break self.connfd.send(data) # 循環接收客戶端請求 def run(self): while True: data = self.connfd.recv(1024).decode() if not data or data == ’Q’:return elif data == ’L’:self.do_list() elif data[0] == ’G’: # G filenamefilename = data.split(’ ’)[-1]self.do_get(filename)# 網絡搭建def main(): # 創建套接字 sockfd = socket() sockfd.setsockopt(SOL_SOCKET,SO_REUSEADDR,1) sockfd.bind(ADDR) sockfd.listen(3) print('Listen the port %d...'%PORT) while True: try: connfd,addr = sockfd.accept() print('Connect from',addr) except KeyboardInterrupt: print('服務器程序退出') return except Exception as e: print(e) continue # 創建新的線程處理客戶端 client = FTPServer(connfd) client.setDaemon(True) client.start() # 運行ｒｕｎ方法if __name__ == '__main__': main()
ftp client：
from socket import *import sysADDR = (’127.0.0.1’,8080) # 服務器地址# 客戶端功能處理類class FTPClient: def __init__(self,sockfd): self.sockfd = sockfd def do_list(self): self.sockfd.send(b’L’) # 發送請求 # 等待回復 data = self.sockfd.recv(128).decode() if data == ’OK’: # 一次接收文件列表字符串 data = self.sockfd.recv(4096) print(data.decode()) else: print(data) def do_get(self,filename): # 發送請求 self.sockfd.send((’G ’+filename).encode()) # 等待回復 data = self.sockfd.recv(128).decode() if data == ’OK’: fd = open(filename,’wb’) # 接收文件 while True:data = self.sockfd.recv(1024)if data == b’##’: breakfd.write(data) fd.close() else: print(data) def do_quit(self): self.sockfd.send(b’Q’) self.sockfd.close() sys.exit('謝謝使用')# 創建客戶端網絡def main(): sockfd = socket() try: sockfd.connect(ADDR) except Exception as e: print(e) return ftp = FTPClient(sockfd) # 實例化對象 # 循環發送請求 while True: print('n=========命令選項==========') print('**** list ****') print('**** get file ****') print('**** put file ****') print('**** quit ****') print('=============================') cmd = input('輸入命令：') if cmd.strip() == ’list’: ftp.do_list() elif cmd[:3] == ’get’: # get filename filename = cmd.strip().split(’ ’)[-1] ftp.do_get(filename) elif cmd[:3] == ’put’: # put ../filename filename = cmd.strip().split(’ ’)[-1] ftp.do_put(filename) elif cmd.strip() == ’quit’: ftp.do_quit() else: print('請輸入正確命令')if __name__ == '__main__': main()五、IO并發
定義：在內存中數據交換的操作被定義為IO操作，IO------輸入輸出
內存和磁盤進行數據交換： 文件的讀寫 數據庫更新
內存和終端數據交換 ：input print sys.stdin sys.stdout sys.stderr
內存和網絡數據的交換： 網絡連接 recv send recvfrom
IO密集型程序 ： 程序執行中有大量的IO操作，而較少的cpu運算操作。消耗cpu較少，IO運行時間長
CPU(計算)密集型程序：程序中存在大量的cpu運算，IO操作相對較少，消耗cpu大。
5.1、IO分類IO分為：阻塞IO、非阻塞IO、IO多路復用、事件驅動IO、異步IO
阻塞IO
定義: 在執行IO操作時如果執行條件不滿足則阻塞。阻塞IO是IO的默認形態。 效率: 阻塞IO是效率很低的一種IO。但是由于邏輯簡單所以是默認IO行為。阻塞情況:
因為某種執行條件沒有滿足造成的函數阻塞e.g. accept input recv 處理IO的時間較長產生的阻塞狀態e.g. 網絡傳輸, 大文件讀寫非阻塞IO
定義 : 通過修改IO屬性行為, 使原本阻塞的IO變為非阻塞的狀態。
設置套接字為非阻塞IO
sockfd.setblocking(bool) 功能: 設置套接字為非阻塞IO 參數: 默認為True,表示套接字IO阻塞;設置為False則套接字IO變為非阻塞超時檢測 :設置一個最長阻塞時間,超過該時間后則不再阻塞等待。
sockfd.settimeout(sec) 功能:設置套接字的超時時間 參數:設置的時間5.2、IO多路復用定義 ：通過一個監測，可以同時監控多個IO事件的行為。當哪個IO事件可以執行，即讓這個IO事件發生。
rs, ws, xs = select(rlist, wlist, xlist[, timeout])監控IO事件，阻塞等待監控的IO時間發生
參數 ：
rlist列表，存放（被動）等待處理的IO （接收） wlist列表，存放主動處理的IO（發送） xlist列表，存放出錯，希望去處理的IO（異常） timeout 超時檢測返回值:
rs列表rlist中準備就緒的IO ws列表wlist中準備就緒的IO xs列表xlist中準備就緒的IOselect 實現tcp服務
1.將關注的IO放入對應的監控類別列表
2.通過select函數進行監控
3.遍歷select返回值列表,確定就緒IO事件
4.處理發生的IO事件
from socket import *from select import select# 創建一個監聽套接字作為關注的ＩＯs = socket()s.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)s.bind((’0.0.0.0’,8888))s.listen(3)# 設置關注列表rlist = [s]wlist = []xlist = [s]# 循環監控ＩＯwhile True: rs,ws,xs = select(rlist,wlist,xlist) # 遍歷三個返回列表,處理ＩＯ for r in rs: # 根據遍歷到ＩＯ的不同使用ｉｆ分情況處理 if r is s: c,addr = r.accept() print('Connect from',addr) rlist.append(c) # 增加新的ＩＯ事件 # else為客戶端套接字就緒情況 else: data = r.recv(1024) # 客戶端退出 if not data:rlist.remove(r) # 從關注列表移除r.close()continue # 繼續處理其他就緒ＩＯ print('Receive:',data.decode()) # r.send(b’OK’) # 我們希望主動處理這個ＩＯ對象 wlist.append(r) for w in ws: w.send(b’OK’) wlist.remove(w) # 使用后移除 for x in xs: pass
注意:
wlist中如果存在IO事件,則select立即返回給ws 處理IO過程中不要出現死循環占有服務端的情況 IO多路復用消耗資源較少,效率較高擴展:5.3、位運算將整數轉換為二進制, 按照二進制位進行運算符操作& 按位與 | 按位或 ^ 按位異或 << 左移 >> 右移11 101114 1110(11 & 14 1010) (11| 14 1111)(11^ 14 0101)11 << 2 ===> 44 右側補014 >> 2 ===> 3 擠掉右側的數字
使用 ：
在做底層硬件時操作寄存器 做標志位的過濾5.4、poll方法實現IO多路復用創建poll對象：p = select.poll()
注冊關注的IO事件：p.register(fd,event)
fd 要關注的IO event 要關注的IO事件類型常用類型:
POLLIN 讀IO事件(rlist) POLLOUT 寫IO事件 (wlist) POLLERR 異常IO (xlist) POLLHUP 斷開連接取消對IO的關注：p.unregister(fd)
參數: IO對象或者IO對象的fileno
events = p.poll()：
功能: 阻塞等待監控的IO事件發生 返回值： 返回發生的IO事件events是一個列表 [(fileno,evnet),(),()....]
每個元組為一個就緒IO,元組第一項是該IO的fileno,第二項為該IO就緒的事件類型
poll_server 步驟
1.創建套接字
2.將套接字register
3.創建查找字典,并維護
4.循環監控IO發生
5.處理發生的IO
from socket import *from select import *# 創建套接字s = socket()s.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)s.bind((’0.0.0.0’,8888))s.listen(3)# 創建ｐｏｌｌ對象關注ｓp = poll()# 建立查找字典，用于通過ｆｉｌｅｎｏ查找ＩＯ對象fdmap = {s.fileno():s}# 關注ｓp.register(s,POLLIN|POLLERR)# 循環監控while True: events = p.poll() # 循環遍歷發生的事件 ｆｄ-->fileno for fd,event in events: # 區分事件進行處理 if fd == s.fileno(): c,addr = fdmap[fd].accept() print('Connect from',addr) # 添加新的關注ＩＯ p.register(c,POLLIN|POLLERR) fdmap[c.fileno()] = c # 維護字典 # 按位與判定是ＰＯＬＬＩＮ就緒 elif event & POLLIN: data = fdmap[fd].recv(1024) if not data:p.unregister(fd) # 取消關注fdmap[fd].close()del fdmap[fd] # 從字典中刪除continue print('Receive:',data.decode()) fdmap[fd].send(b’OK’)5.5、epoll方法
1. 使用方法 : 基本與poll相同
生成對象改為 epoll() 將所有事件類型改為EPOLL類型2. epoll特點
epoll 效率比select poll要高 epoll 監控IO數量比select要多 epoll 的觸發方式比poll要多 (EPOLLET邊緣觸發)from socket import *from select import *# 創建套接字s = socket()s.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)s.bind((’0.0.0.0’,8888))s.listen(3)# 創建eｐｏｌｌ對象關注ｓep = epoll()# 建立查找字典，用于通過ｆｉｌｅｎｏ查找ＩＯ對象fdmap = {s.fileno():s}# 關注ｓep.register(s,EPOLLIN|EPOLLERR)# 循環監控while True: events = ep.poll() # 循環遍歷發生的事件 ｆｄ-->fileno for fd,event in events: print('親，你有ＩＯ需要處理哦') # 區分事件進行處理 if fd == s.fileno(): c,addr = fdmap[fd].accept() print('Connect from',addr) # 添加新的關注ＩＯ # 將觸發方式變為邊緣觸發 ep.register(c,EPOLLIN|EPOLLERR|EPOLLET) fdmap[c.fileno()] = c # 維護字典 # 按位與判定是EＰＯＬＬＩＮ就緒 # elif event & EPOLLIN: # data = fdmap[fd].recv(1024) # if not data: # ep.unregister(fd) # 取消關注 # fdmap[fd].close() # del fdmap[fd] # 從字典中刪除 # continue # print('Receive:',data.decode()) # fdmap[fd].send(b’OK’)
以上就是分析python并發網絡通信模型的詳細內容，更多關于python 并發網絡通信模型的資料請關注好吧啦網其它相關文章！
相關文章：
網公網安備