目錄一、常見模型分類1.1、循環(huán)服務(wù)器模型1.2、IO并發(fā)模型1.3、多進(jìn)程/線程網(wǎng)絡(luò)并發(fā)模型二、基于fork的多進(jìn)程網(wǎng)絡(luò)并發(fā)模型三、基于threading的多線程網(wǎng)絡(luò)并發(fā)四、ftp 文件服務(wù)器4.1、項(xiàng)目功能4.2、整體結(jié)構(gòu)設(shè)計(jì)五、IO并發(fā)5.1、IO分類5.2、IO多路復(fù)用5.3、位運(yùn)算5.4、poll方法實(shí)現(xiàn)IO多路復(fù)用5.5、epoll方法一、常見模型分類1.1、循環(huán)服務(wù)器模型

循環(huán)接收客戶端請(qǐng)求,處理請(qǐng)求。同一時(shí)刻只能處理一個(gè)請(qǐng)求,處理完畢后再處理下一個(gè)。

優(yōu)點(diǎn):實(shí)現(xiàn)簡(jiǎn)單,占用資源少 缺點(diǎn):無法同時(shí)處理多個(gè)客戶端請(qǐng)求 適用情況:處理的任務(wù)可以很快完成,客戶端無需長(zhǎng)期占用服務(wù)端程序。udp比tcp更適合循環(huán)。1.2、IO并發(fā)模型

利用IO多路復(fù)用,異步IO等技術(shù),同時(shí)處理多個(gè)客戶端IO請(qǐng)求。

優(yōu)點(diǎn) : 資源消耗少,能同時(shí)高效處理多個(gè)IO行為 缺點(diǎn) : 只能處理并發(fā)產(chǎn)生的IO事件,無法處理cpu計(jì)算 適用情況:HTTP請(qǐng)求,網(wǎng)絡(luò)傳輸?shù)榷际荌O行為。1.3、多進(jìn)程/線程網(wǎng)絡(luò)并發(fā)模型

每當(dāng)一個(gè)客戶端連接服務(wù)器,就創(chuàng)建一個(gè)新的進(jìn)程/線程為該客戶端服務(wù),客戶端退出時(shí)再銷毀該進(jìn)程/線程。

優(yōu)點(diǎn):能同時(shí)滿足多個(gè)客戶端長(zhǎng)期占有服務(wù)端需求,可以處理各種請(qǐng)求。 缺點(diǎn): 資源消耗較大 適用情況:客戶端同時(shí)連接量較少,需要處理行為較復(fù)雜情況。二、基于fork的多進(jìn)程網(wǎng)絡(luò)并發(fā)模型

1.創(chuàng)建監(jiān)聽套接字

2.等待接收客戶端請(qǐng)求

3.客戶端連接創(chuàng)建新的進(jìn)程處理客戶端請(qǐng)求

4.原進(jìn)程繼續(xù)等待其他客戶端連接

5.如果客戶端退出,則銷毀對(duì)應(yīng)的進(jìn)程

from socket import *import osimport signal# 創(chuàng)建監(jiān)聽套接字HOST = ’0.0.0.0’PORT = 8888ADDR = (HOST,PORT)# 客戶端服務(wù)函數(shù)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) # 設(shè)置套接字端口重用s.bind(ADDR)s.listen(3)signal.signal(signal.SIGCHLD,signal.SIG_IGN) # 處理僵尸進(jìn)程print('Listen the port %d...' % PORT)# 循環(huán)等待客戶端連接while True: try: c,addr = s.accept() except KeyboardInterrupt: os._exit(0) except Exception as e: print(e) continue # 創(chuàng)建子進(jìn)程處理這個(gè)客戶端 pid = os.fork() if pid == 0: # 處理客戶端請(qǐng)求 s.close() handle(c) os._exit(0) # handle處理完客戶端請(qǐng)求子進(jìn)程也退出 # 無論出錯(cuò)或者父進(jìn)程都要循環(huán)回去接受請(qǐng)求 # c對(duì)于父進(jìn)程沒用 c.close()三、基于threading的多線程網(wǎng)絡(luò)并發(fā)

1.創(chuàng)建監(jiān)聽套接字

2.循環(huán)接收客戶端連接請(qǐng)求

3.當(dāng)有新的客戶端連接創(chuàng)建線程處理客戶端請(qǐng)求

4.主線程繼續(xù)等待其他客戶端連接

5.當(dāng)客戶端退出,則對(duì)應(yīng)分支線程退出

from socket import *from threading import Threadimport sys# 創(chuàng)建監(jiān)聽套接字HOST = ’0.0.0.0’PORT = 8888ADDR = (HOST,PORT)# 處理客戶端請(qǐng)求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)# 循環(huán)等待客戶端連接while True: try: c,addr = s.accept() except KeyboardInterrupt: sys.exit('服務(wù)器退出') except Exception as e: print(e) continue # 創(chuàng)建線程處理客戶端請(qǐng)求 t = Thread(target=handle, args=(c,)) t.setDaemon(True) # 父進(jìn)程結(jié)束則所有進(jìn)程終止 t.start()四、ftp 文件服務(wù)器4.1、項(xiàng)目功能

客戶端有簡(jiǎn)單的頁面命令提示:功能包含：

查看服務(wù)器文件庫(kù)中的文件列表（普通文件） 可以下載其中的某個(gè)文件到本地 可以上傳客戶端文件到服務(wù)器文件庫(kù)

服務(wù)器需求 ：

允許多個(gè)客戶端同時(shí)操作 每個(gè)客戶端可能回連續(xù)發(fā)送命令

技術(shù)分析：

tcp套接字更適合文件傳輸 并發(fā)方案 ---》 fork 多進(jìn)程并發(fā) 對(duì)文件的讀寫操作獲取 文件列表 ----》 os.listdir()

粘包的處理

4.2、整體結(jié)構(gòu)設(shè)計(jì) 服務(wù)器功能封裝在類中（上傳，下載，查看列表） 創(chuàng)建套接字，流程函數(shù)調(diào)用 main（） 客戶端負(fù)責(zé)發(fā)起請(qǐng)求，接受回復(fù)，展示 服務(wù)端負(fù)責(zé)接受請(qǐng)求，邏輯處理

ftp server：

from socket import *from threading import Threadimport osimport time#　全局變量HOST = ’0.0.0.0’PORT = 8080ADDR = (HOST,PORT)FTP = '/home/tarena/FTP/' #　文件庫(kù)位置# 創(chuàng)建文件服務(wù)器服務(wù)端功能類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('文件庫(kù)為空'.encode()) return else: self.connfd.send(b’OK’) time.sleep(0.1) #　防止和后面發(fā)送內(nèi)容粘包 #　拼接文件列表 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) #　文件發(fā)送 while True: data = fd.read(1024) if not data:time.sleep(0.1)self.connfd.send(b’##’)break self.connfd.send(data) #　循環(huán)接收客戶端請(qǐng)求 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)# 網(wǎng)絡(luò)搭建def main(): # 創(chuàng)建套接字 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('服務(wù)器程序退出') return except Exception as e: print(e) continue #　創(chuàng)建新的線程處理客戶端 client = FTPServer(connfd) client.setDaemon(True) client.start() #　運(yùn)行ｒｕｎ方法if __name__ == '__main__': main()

ftp client：

from socket import *import sysADDR = (’127.0.0.1’,8080) # 服務(wù)器地址#　客戶端功能處理類class FTPClient: def __init__(self,sockfd): self.sockfd = sockfd def do_list(self): self.sockfd.send(b’L’) #　發(fā)送請(qǐng)求 # 等待回復(fù) 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): #　發(fā)送請(qǐng)求 self.sockfd.send((’G ’+filename).encode()) # 等待回復(fù) 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('謝謝使用')#　創(chuàng)建客戶端網(wǎng)絡(luò)def main(): sockfd = socket() try: sockfd.connect(ADDR) except Exception as e: print(e) return ftp = FTPClient(sockfd) #　實(shí)例化對(duì)象 #　循環(huán)發(fā)送請(qǐng)求 while True: print('n=========命令選項(xiàng)==========') 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('請(qǐng)輸入正確命令')if __name__ == '__main__': main()五、IO并發(fā)

定義：在內(nèi)存中數(shù)據(jù)交換的操作被定義為IO操作，IO------輸入輸出

內(nèi)存和磁盤進(jìn)行數(shù)據(jù)交換： 文件的讀寫 數(shù)據(jù)庫(kù)更新

內(nèi)存和終端數(shù)據(jù)交換 ：input print sys.stdin sys.stdout sys.stderr

內(nèi)存和網(wǎng)絡(luò)數(shù)據(jù)的交換： 網(wǎng)絡(luò)連接 recv send recvfrom

IO密集型程序 ： 程序執(zhí)行中有大量的IO操作，而較少的cpu運(yùn)算操作。消耗cpu較少，IO運(yùn)行時(shí)間長(zhǎng)

CPU(計(jì)算)密集型程序：程序中存在大量的cpu運(yùn)算，IO操作相對(duì)較少，消耗cpu大。

5.1、IO分類

IO分為：阻塞IO、非阻塞IO、IO多路復(fù)用、事件驅(qū)動(dòng)IO、異步IO

阻塞IO

定義: 在執(zhí)行IO操作時(shí)如果執(zhí)行條件不滿足則阻塞。阻塞IO是IO的默認(rèn)形態(tài)。 效率: 阻塞IO是效率很低的一種IO。但是由于邏輯簡(jiǎn)單所以是默認(rèn)IO行為。

阻塞情況:

因?yàn)槟撤N執(zhí)行條件沒有滿足造成的函數(shù)阻塞e.g. accept input recv 處理IO的時(shí)間較長(zhǎng)產(chǎn)生的阻塞狀態(tài)e.g. 網(wǎng)絡(luò)傳輸, 大文件讀寫

非阻塞IO

定義 : 通過修改IO屬性行為, 使原本阻塞的IO變?yōu)榉亲枞臓顟B(tài)。

設(shè)置套接字為非阻塞IO

sockfd.setblocking(bool) 功能: 設(shè)置套接字為非阻塞IO 參數(shù): 默認(rèn)為True,表示套接字IO阻塞;設(shè)置為False則套接字IO變?yōu)榉亲枞?p>超時(shí)檢測(cè) :設(shè)置一個(gè)最長(zhǎng)阻塞時(shí)間,超過該時(shí)間后則不再阻塞等待。

sockfd.settimeout(sec) 功能:設(shè)置套接字的超時(shí)時(shí)間 參數(shù):設(shè)置的時(shí)間5.2、IO多路復(fù)用

定義 ：通過一個(gè)監(jiān)測(cè)，可以同時(shí)監(jiān)控多個(gè)IO事件的行為。當(dāng)哪個(gè)IO事件可以執(zhí)行，即讓這個(gè)IO事件發(fā)生。

rs, ws, xs = select(rlist, wlist, xlist[, timeout])監(jiān)控IO事件，阻塞等待監(jiān)控的IO時(shí)間發(fā)生

參數(shù) ：

rlist列表，存放（被動(dòng)）等待處理的IO （接收） wlist列表，存放主動(dòng)處理的IO（發(fā)送） xlist列表，存放出錯(cuò)，希望去處理的IO（異常） timeout 超時(shí)檢測(cè)

返回值:

rs列表rlist中準(zhǔn)備就緒的IO ws列表wlist中準(zhǔn)備就緒的IO xs列表xlist中準(zhǔn)備就緒的IO

select 實(shí)現(xiàn)tcp服務(wù)

1.將關(guān)注的IO放入對(duì)應(yīng)的監(jiān)控類別列表

2.通過select函數(shù)進(jìn)行監(jiān)控

3.遍歷select返回值列表,確定就緒IO事件

4.處理發(fā)生的IO事件

from socket import *from select import select#　創(chuàng)建一個(gè)監(jiān)聽套接字作為關(guān)注的ＩＯs = socket()s.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)s.bind((’0.0.0.0’,8888))s.listen(3)#　設(shè)置關(guān)注列表rlist = [s]wlist = []xlist = [s]#　循環(huán)監(jiān)控ＩＯwhile True: rs,ws,xs = select(rlist,wlist,xlist) # 遍歷三個(gè)返回列表,處理ＩＯ for r in rs: # 根據(jù)遍歷到ＩＯ的不同使用ｉｆ分情況處理 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) #　從關(guān)注列表移除r.close()continue # 繼續(xù)處理其他就緒ＩＯ print('Receive:',data.decode()) # r.send(b’OK’) #　我們希望主動(dòng)處理這個(gè)ＩＯ對(duì)象 wlist.append(r) for w in ws: w.send(b’OK’) wlist.remove(w) #　使用后移除 for x in xs: pass

注意:

wlist中如果存在IO事件,則select立即返回給ws 處理IO過程中不要出現(xiàn)死循環(huán)占有服務(wù)端的情況 IO多路復(fù)用消耗資源較少,效率較高擴(kuò)展:5.3、位運(yùn)算

將整數(shù)轉(zhuǎn)換為二進(jìn)制, 按照二進(jìn)制位進(jìn)行運(yùn)算符操作& 按位與 | 按位或 ^ 按位異或 << 左移 >> 右移11 101114 1110(11 & 14 1010) (11| 14 1111)(11^ 14 0101)11 << 2 ===> 44 右側(cè)補(bǔ)014 >> 2 ===> 3 擠掉右側(cè)的數(shù)字

使用 ：

在做底層硬件時(shí)操作寄存器 做標(biāo)志位的過濾5.4、poll方法實(shí)現(xiàn)IO多路復(fù)用

創(chuàng)建poll對(duì)象：p = select.poll()

注冊(cè)關(guān)注的IO事件：p.register(fd,event)

fd 要關(guān)注的IO event 要關(guān)注的IO事件類型

常用類型:

POLLIN 讀IO事件(rlist) POLLOUT 寫IO事件 (wlist) POLLERR 異常IO (xlist) POLLHUP 斷開連接

取消對(duì)IO的關(guān)注：p.unregister(fd)

參數(shù): IO對(duì)象或者IO對(duì)象的fileno

events = p.poll()：

功能: 　 阻塞等待監(jiān)控的IO事件發(fā)生 返回值： 返回發(fā)生的IO事件

events是一個(gè)列表 [(fileno,evnet),(),()....]

每個(gè)元組為一個(gè)就緒IO,元組第一項(xiàng)是該IO的fileno,第二項(xiàng)為該IO就緒的事件類型

poll_server 步驟

1.創(chuàng)建套接字

2.將套接字register

3.創(chuàng)建查找字典,并維護(hù)

4.循環(huán)監(jiān)控IO發(fā)生

5.處理發(fā)生的IO

from socket import *from select import *#　創(chuàng)建套接字s = socket()s.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)s.bind((’0.0.0.0’,8888))s.listen(3)# 創(chuàng)建ｐｏｌｌ對(duì)象關(guān)注ｓp = poll()#　建立查找字典，用于通過ｆｉｌｅｎｏ查找ＩＯ對(duì)象fdmap = {s.fileno():s}# 關(guān)注ｓp.register(s,POLLIN|POLLERR)#　循環(huán)監(jiān)控while True: events = p.poll() #　循環(huán)遍歷發(fā)生的事件　ｆｄ-->fileno for fd,event in events: #　區(qū)分事件進(jìn)行處理 if fd == s.fileno(): c,addr = fdmap[fd].accept() print('Connect from',addr) #　添加新的關(guān)注ＩＯ p.register(c,POLLIN|POLLERR) fdmap[c.fileno()] = c #　維護(hù)字典 #　按位與判定是ＰＯＬＬＩＮ就緒 elif event & POLLIN: data = fdmap[fd].recv(1024) if not data:p.unregister(fd) #　取消關(guān)注fdmap[fd].close()del fdmap[fd] #　從字典中刪除continue print('Receive:',data.decode()) fdmap[fd].send(b’OK’)5.5、epoll方法

1. 使用方法 : 基本與poll相同

生成對(duì)象改為 epoll() 將所有事件類型改為EPOLL類型

2. epoll特點(diǎn)

epoll 效率比select poll要高 epoll 監(jiān)控IO數(shù)量比select要多 epoll 的觸發(fā)方式比poll要多 (EPOLLET邊緣觸發(fā))

from socket import *from select import *#　創(chuàng)建套接字s = socket()s.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)s.bind((’0.0.0.0’,8888))s.listen(3)# 創(chuàng)建eｐｏｌｌ對(duì)象關(guān)注ｓep = epoll()#　建立查找字典，用于通過ｆｉｌｅｎｏ查找ＩＯ對(duì)象fdmap = {s.fileno():s}# 關(guān)注ｓep.register(s,EPOLLIN|EPOLLERR)#　循環(huán)監(jiān)控while True: events = ep.poll() #　循環(huán)遍歷發(fā)生的事件　ｆｄ-->fileno for fd,event in events: print('親，你有ＩＯ需要處理哦') #　區(qū)分事件進(jìn)行處理 if fd == s.fileno(): c,addr = fdmap[fd].accept() print('Connect from',addr) #　添加新的關(guān)注ＩＯ #　將觸發(fā)方式變?yōu)檫吘売|發(fā) ep.register(c,EPOLLIN|EPOLLERR|EPOLLET) fdmap[c.fileno()] = c #　維護(hù)字典 #　按位與判定是EＰＯＬＬＩＮ就緒 # elif event & EPOLLIN: # data = fdmap[fd].recv(1024) # if not data: # ep.unregister(fd) #　取消關(guān)注 # fdmap[fd].close() # del fdmap[fd] #　從字典中刪除 # continue # print('Receive:',data.decode()) # fdmap[fd].send(b’OK’)

以上就是分析python并發(fā)網(wǎng)絡(luò)通信模型的詳細(xì)內(nèi)容，更多關(guān)于python 并發(fā)網(wǎng)絡(luò)通信模型的資料請(qǐng)關(guān)注好吧啦網(wǎng)其它相關(guān)文章！