分析python并發(fā)網(wǎng)絡(luò)通信模型

分析python并發(fā)?絡(luò)通信模型?、常見模型分類1.1、循環(huán)服務(wù)器模型循環(huán)接收客戶端請求,處理請求。同?時刻只能處理?個請求,處理完畢后再處理下?個。優(yōu)點:實現(xiàn)簡單,占?資源少缺點:?法同時處理多個客戶端請求適?情況:處理的任務(wù)可以很快完成,客戶端?需長期占?服務(wù)端程序。udp?tcp更適合循環(huán)。1.2、IO并發(fā)模型利?IO多路復(fù)?,異步IO等技術(shù),同時處理多個客戶端IO請求。優(yōu)點:資源消耗少,能同時?效處理多個IO?為缺點:只能處理并發(fā)產(chǎn)?的IO事件,?法處理cpu計算適?情況:HTTP請求,?絡(luò)傳輸?shù)榷际荌O?為。1.3、多進程/線程?絡(luò)并發(fā)模型每當?個客戶端連接服務(wù)器,就創(chuàng)建?個新的進程/線程為該客戶端服務(wù),客戶端退出時再銷毀該進程/線程。優(yōu)點:能同時滿?多個客戶端長期占有服務(wù)端需求,可以處理各種請求。缺點:資源消耗較?適?情況:客戶端同時連接量較少,需要處理?為較復(fù)雜情況。?、基于fork的多進程?絡(luò)并發(fā)模型1.創(chuàng)建監(jiān)聽套接字2.等待接收客戶端請求3.客戶端連接創(chuàng)建新的進程處理客戶端請求4.原進程繼續(xù)等待其他客戶端連接5.如果客戶端退出,則銷毀對應(yīng)的進程fromsocketimport*1importos2importsignal34#創(chuàng)建監(jiān)聽套接字5HOST=''6PORT=88887ADDR=(HOST,PORT)89#客戶端服務(wù)函數(shù)10defhandle(c):11whileTrue:12data=c.recv(1024)1314151617ifnotdata:breakprint(data.decode())c.send(b'OK')1718192021222324252627282930313233343536373839404142434445464748c.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)#處理僵?進程print("Listentheport%d..."%PORT)#循環(huán)等待客戶端連接whileTrue:try:c,addr=s.accept()exceptKeyboardInterrupt:os._exit(0)exceptExceptionase:print(e)continue#創(chuàng)建?進程處理這個客戶端pid=os.fork()ifpid==0:#處理客戶端請求s.close()handle(c)os._exit(0)#handle處理完客戶端請求?進程也退出#?論出錯或者?進程都要循環(huán)回去接受請求#c對于?進程沒?c.close()三、基于threading的多線程?絡(luò)并發(fā)1.創(chuàng)建監(jiān)聽套接字2.循環(huán)接收客戶端連接請求3.當有新的客戶端連接創(chuàng)建線程處理客戶端請求4.主線程繼續(xù)等待其他客戶端連接5.當客戶端退出,則對應(yīng)分?線程退出fromsocketimport*1fromthreadingimportThreadimportsys23#創(chuàng)建監(jiān)聽套接字45HOST=''PORT=88886ADDR=(HOST,PORT)78#處理客戶端請求defhandle(c):whileTrue:910111112131415161718192021222324252627282930313233343536373839whileTrue:data=c.recv(1024)ifnotdata:breakprint(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("Listentheport%d..."%PORT)#循環(huán)等待客戶端連接whileTrue:try:c,addr=s.accept()exceptKeyboardInterrupt:sys.exit("服務(wù)器退出")exceptExceptionase:print(e)continue#創(chuàng)建線程處理客戶端請求t=Thread(target=handle,args=(c,))t.setDaemon(True)#?進程結(jié)束則所有進程終?t.start()四、ftp?件服務(wù)器4.1、項?功能客戶端有簡單的頁?命令提?:功能包含：查看服務(wù)器?件庫中的?件列表（普通?件）可以下載其中的某個?件到本地可以上傳客戶端?件到服務(wù)器?件庫服務(wù)器需求：允許多個客戶端同時操作每個客戶端可能回連續(xù)發(fā)送命令技術(shù)分析：tcp套接字更適合?件傳輸并發(fā)?案---》fork多進程并發(fā)對?件的讀寫操作獲取?件列表----》os.listdir()粘包的處理4.2、整體結(jié)構(gòu)設(shè)計服務(wù)器功能封裝在類中（上傳，下載，查看列表）創(chuàng)建套接字，流程函數(shù)調(diào)?main（）客戶端負責(zé)發(fā)起請求，接受回復(fù)，展?服務(wù)端負責(zé)接受請求，邏輯處理ftpserver：fromsocketimport*fromthreadingimportThreadimportos123importtime45#全局變量6HOST=''7PORT=80808ADDR=(HOST,PORT)FTP="/home/tarena/FTP/"#?件庫位置910111213141516171819202122232425262728293031323334353637383940414243444546#創(chuàng)建?件服務(wù)器服務(wù)端功能類classFTPServer(Thread):def__init__(self,connfd):self.connfd=connfdsuper().__init__()defdo_list(self):#獲取?件列表files=os.listdir(FTP)ifnotfiles:self.connfd.send("?件庫為空".encode())returnelse:self.connfd.send(b'OK')time.sleep(0.1)#防?和后?發(fā)送內(nèi)容粘包#拼接?件列表files_=""forfileinfiles:iffile[0]!='.'and\os.path.isfile(FTP+file):files_+=file+'\n'self.connfd.send(files_.encode())defdo_get(self,filename):try:fd=open(FTP+filename,'rb')exceptException:self.connfd.send("?件不存在".encode())returnelse:self.connfd.send(b'OK')time.sleep(0.1)#?件發(fā)送whileTrue:data=fd.read(1024)ifnotdata:47484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192time.sleep(0.1)self.connfd.send(b'##')breakself.connfd.send(data)#循環(huán)接收客戶端請求defrun(self):whileTrue:data=self.connfd.recv(1024).decode()ifnotdataordata=='Q':returnelifdata=='L':self.do_list()elifdata[0]=='G':#Gfilenamefilename=data.split('')[-1]self.do_get(filename)#?絡(luò)搭建defmain():#創(chuàng)建套接字sockfd=socket()sockfd.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)sockfd.bind(ADDR)sockfd.listen(3)print("Listentheport%d..."%PORT)whileTrue:try:connfd,addr=sockfd.accept()print("Connectfrom",addr)exceptKeyboardInterrupt:print("服務(wù)器程序退出")returnexceptExceptionase:print(e)continue#創(chuàng)建新的線程處理客戶端client=FTPServer(connfd)client.setDaemon(True)client.start()#運?ｒｕｎ?法if__name__=="__main__":main()ftpclient：fromsocketimport*importsys123ADDR=('',8080)#服務(wù)器地址4#客戶端功能處理類classFTPClient:567def__init__(self,sockfd):self.sockfd=sockfd89defdo_list(self):10111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758self.sockfd.send(b'L')#發(fā)送請求#等待回復(fù)data=self.sockfd.recv(128).decode()ifdata=='OK':#?次接收?件列表字符串data=self.sockfd.recv(4096)print(data.decode())else:print(data)defdo_get(self,filename):#發(fā)送請求self.sockfd.send(('G'+filename).encode())#等待回復(fù)data=self.sockfd.recv(128).decode()ifdata=='OK':fd=open(filename,'wb')#接收?件whileTrue:data=self.sockfd.recv(1024)ifdata==b'##':breakfd.write(data)fd.close()else:print(data)defdo_quit(self):self.sockfd.send(b'Q')self.sockfd.close()sys.exit("謝謝使?")#創(chuàng)建客戶端?絡(luò)defmain():sockfd=socket()try:sockfd.connect(ADDR)exceptExceptionase:print(e)returnftp=FTPClient(sockfd)#實例化對象#循環(huán)發(fā)送請求whileTrue:print("\n=========命令選項==========")print("****list****")print("****getfileprint("****putfile****")****")****")5960616263646566676869707172737475767778798081828384print("****quitprint("=============================")cmd=input("輸?命令：")ifcmd.strip()=='list':ftp.do_list()elifcmd[:3]=='get':#getfilenamefilename=cmd.strip().split('')[-1]ftp.do_get(filename)elifcmd[:3]=='put':#put../filenamefilename=cmd.strip().split('')[-1]ftp.do_put(filename)elifcmd.strip()=='quit':ftp.do_quit()else:print("請輸?正確命令")if__name__=="__main__":main()五、IO并發(fā)定義：在內(nèi)存中數(shù)據(jù)交換的操作被定義為IO操作，IO------輸?輸出內(nèi)存和磁盤進?數(shù)據(jù)交換：?件的讀寫數(shù)據(jù)庫更新內(nèi)存和終端數(shù)據(jù)交換：inputprintsys.stdinsys.stdoutsys.stderr內(nèi)存和?絡(luò)數(shù)據(jù)的交換：?絡(luò)連接recvsendrecvfromIO密集型程序：程序執(zhí)?中有?量的IO操作，?較少的cpu運算操作。消耗cpu較少，IO運?時間長CPU(計算)密集型程序：程序中存在?量的cpu運算，IO操作相對較少，消耗cpu?。5.1、IO分類IO分為：阻塞IO、?阻塞IO、IO多路復(fù)?、事件驅(qū)動IO、異步IO阻塞IO定義:在執(zhí)?IO操作時如果執(zhí)?條件不滿?則阻塞。阻塞IO是IO的默認形態(tài)。效率:阻塞IO是效率很低的?種IO。但是由于邏輯簡單所以是默認IO?為。阻塞情況:因為某種執(zhí)?條件沒有滿?造成的函數(shù)阻塞e.g.acceptinputrecv處理IO的時間較長產(chǎn)?的阻塞狀態(tài)e.g.?絡(luò)傳輸,??件讀寫?阻塞IO定義:通過修改IO屬性?為,使原本阻塞的IO變?yōu)?阻塞的狀態(tài)。設(shè)置套接字為?阻塞IOsockfd.setblocking(bool)功能:設(shè)置套接字為?阻塞IO參數(shù):默認為True,表?套接字IO阻塞;設(shè)置為False則套接字IO變?yōu)?阻塞超時檢測:設(shè)置?個最長阻塞時間,超過該時間后則不再阻塞等待。sockfd.settimeout(sec)功能:設(shè)置套接字的超時時間參數(shù):設(shè)置的時間5.2、IO多路復(fù)?定義：通過?個監(jiān)測，可以同時監(jiān)控多個IO事件的?為。當哪個IO事件可以執(zhí)?，即讓這個IO事件發(fā)?。rs,ws,xs=select(rlist,wlist,xlist[,timeout])監(jiān)控IO事件，阻塞等待監(jiān)控的IO時間發(fā)?參數(shù)：rlist列表，存放（被動）等待處理的IO（接收）wlist列表，存放主動處理的IO（發(fā)送）xlist列表，存放出錯，希望去處理的IO（異常）timeout超時檢測返回值:rs列表rlist中準備就緒的IOws列表wlist中準備就緒的IOxs列表xlist中準備就緒的IOselect實現(xiàn)tcp服務(wù)1.將關(guān)注的IO放?對應(yīng)的監(jiān)控類別列表2.通過select函數(shù)進?監(jiān)控3.遍歷select返回值列表,確定就緒IO事件4.處理發(fā)?的IO事件fromsocketimport*1fromselectimportselect23#創(chuàng)建?個監(jiān)聽套接字作為關(guān)注的ＩＯ4s=socket()5s.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)6s.bind(('',8888))7s.listen(3)89#設(shè)置關(guān)注列表10rlist=[s]11wlist=[]12xlist=[s]13141516171819202122232425262728293031323334353637383940414243#循環(huán)監(jiān)控ＩＯwhileTrue:rs,ws,xs=select(rlist,wlist,xlist)#遍歷三個返回列表,處理ＩＯforrinrs:#根據(jù)遍歷到ＩＯ的不同使?ｉｆ分情況處理ifriss:c,addr=r.accept()print("Connectfrom",addr)rlist.append(c)#增加新的ＩＯ事件#else為客戶端套接字就緒情況else:data=r.recv(1024)#客戶端退出ifnotdata:rlist.remove(r)#從關(guān)注列表移除r.close()continue#繼續(xù)處理其他就緒ＩＯprint("Receive:",data.decode())#r.send(b'OK')#我們希望主動處理這個ＩＯ對象wlist.append(r)forwinws:w.send(b'OK')wlist.remove(w)#使?后移除forxinxs:pass注意:wlist中如果存在IO事件,則select?即返回給ws處理IO過程中不要出現(xiàn)死循環(huán)占有服務(wù)端的情況IO多路復(fù)?消耗資源較少,效率較?擴展:5.3、位運算將整數(shù)轉(zhuǎn)換為?進制,按照?進制位進?運算符操作&按位與|按位或^按位異或<<左移>>右移111011141110(11&141010)(11|141111)(11^140101)11<<2===>44右側(cè)補014>>2===>3擠掉右側(cè)的數(shù)字使?：在做底層硬件時操作寄存器做標志位的過濾5.4、poll?法實現(xiàn)IO多路復(fù)?創(chuàng)建poll對象：p=select.poll()注冊關(guān)注的IO事件：p.register(fd,event)fd要關(guān)注的IOevent要關(guān)注的IO事件類型常?類型:POLLIN讀IO事件(rlist)POLLOUT寫IO事件(wlist)POLLERR異常IO(xlist)POLLHUP斷開連接取消對IO的關(guān)注：p.unregister(fd)參數(shù):IO對象或者IO對象的filenoevents=p.poll()：功能:阻塞等待監(jiān)控的IO事件發(fā)?返回值：返回發(fā)?的IO事件events是?個列表[(fileno,evnet),(),()....]每個元組為?個就緒IO,元組第?項是該IO的fileno,第?項為該IO就緒的事件類型poll_server步驟1.創(chuàng)建套接字2.將套接字register3.創(chuàng)建查找字典,并維護4.循環(huán)監(jiān)控IO發(fā)?5.處理發(fā)?的IOfromsocketimport*1fromselectimport*23#創(chuàng)建套接字4s=socket()5s.setsockopt(SOL_SOCKET,SO_REUSEADDR,1)6s.bind(('',8888))7s.listen(3)89#創(chuàng)建ｐｏｌｌ對象關(guān)注ｓ10p=poll()1112#建?查找字典，?于通過ｆｉｌｅｎｏ查找ＩＯ對象13fdmap={s.fileno():s}1415#關(guān)注ｓ16p.register(s,POLLIN|POLLERR)1718#循環(huán)監(jiān)控19whileTrue:20events=p.poll()21#循環(huán)遍歷發(fā)?的事件ｆｄ-->fileno2223forfd,eventinevents:#區(qū)分事件進?處理iffd==s.fileno():2425262728293031323334353637383940c,addr=fdmap[fd].accept()print("Connectfrom",addr)#添加新的關(guān)注ＩＯp.register(c,POLLIN|POLLERR)fdmap[c.fileno()]=c#維護字典#按位與判定是ＰＯＬＬＩＮ就緒elifevent&POLLIN:data=fdmap[fd].recv(1024)ifnotdata:p.unregister(fd)#取消關(guān)注fdmap[fd].close()delfdmap[fd]#從字典中刪除continueprint("Receive:",data.decode())fdmap[fd].send(b'OK')5.5、epoll?法1.使??法:基本與po