Python中的select、epoll詳解

文章目錄

• Python中的select、epoll詳解
• • 一、select
  • • 1、相關概念
    • 2.select的特性
    • • 1.那么單進程是如何實現多并發的呢？？？
      • 2.select的原理
      • 3.select 優點
      • 4.select 缺點
      • 5.python select
      • 6.select 示例:
  • 二、poll
  • • 1.相關概念
    • 2.poll的原理
    • 3.代碼
  • 三、epoll
  • • 1.相關概念
    • 2、epoll原理
    • 3.優缺點
    • • select
      • poll
      • 3.epoll
      • 4. FD劇增后帶來的IO效率問題
      • 5.消息傳遞方式
    • 4.在Python中調用epoll
    • 5.例
  • 四、web靜態服務器-epool

一、select

1、相關概念

• select是通過一個select()系統調用來監視多個文件描述符的數組(在linux中一切事物皆文件，塊設備，socket連接等)
• 當select()返回后，該數組中就緒的文件描述符便會被內核修改標志位（變成ready）
• 使得進程可以獲得這些文件描述符從而進行后續的讀寫操作
• select會不斷監視網絡接口的某個目錄下有多少文件描述符變成ready狀態。在網絡接口中，過來一個連接就會建立一個’文件’
• 變成ready狀態后，select就可以操作這個文件描述符了。

為什么要用一個進程實現多并發而不采用多線程實現多并發呢？
答：因為一個進程實現多并發比多線程是實現多并發的效率還要高，因為啟動多線程會有很多的開銷，而且CPU要不斷的檢查每個線程的狀態，確定哪個線程是否可以執行。這個對系統來說也是有壓力的，用單進程的話就可以避免這種開銷和給系統帶來的壓力，

2.select的特性

1.那么單進程是如何實現多并發的呢？？？

答：

• 很巧妙的使用了生產者和消費者的模式（異步）
• 生產者和消費者可以實現非阻塞，一個socketserver通過select接收多個連接過來（之前的socket一個進程只能接收一個連接，當接收新的連接的時候產生阻塞，因為這個socket進程要先和客戶端進行通信，二者是彼此互相阻塞等待的
• 這個時候如果再來一個連接，要等之前的那個連接斷了，這個才可以連進來。也就是說用基本的socket實現多進程是阻塞的。
• 為了解決這個問題采用每來一個連接產生一個線程，是不阻塞了，但是當線程數量過多的時候，對于cpu來說開銷和壓力是比較大的。
• 對于單個socket來說，阻塞的時候大部分的時候都是在等待IO操作(網絡操作也屬于IO操作)。為了避免這種情況，就出現了異步。
• 客戶端發起一個連接，會在服務端注冊一個文件句柄，服務端會不斷輪詢這些文件句柄的列表
• 主進程和客戶端建立連接而沒有啟動線程，這個時候主進程和客戶端進行交互，其他的客戶端是無法連接主進程的
• 為了實現主進程既能和已連接的客戶端收發消息，又能和新的客戶端建立連接，就把輪詢變的非常快（死循環）去刷客戶端連接進來的文件句柄的列表
• 只要客戶端發消息了，服務端讀取了消息之后，有另一個列表去接收給客戶端返回的消息，也不斷的去刷這個列表，刷出來后返回給客戶端，這樣和客戶端的這次通信就完成了，但是跟客戶端的連接還沒有斷，但是就進入了下一次的輪詢。】

2.select的原理

• 1.從用戶空間拷貝fd_set到內核空間（fd_set 過大導致占用空間且慢）；
• 2.注冊回調函數__pollwait；
• 3.遍歷所有fd，對全部指定設備做一次poll（這里的poll是一個文件操作，它有兩個參數，一個是文件fd本身，一個是當設備尚未就緒時調用的回調函數__pollwait，這個函數把設備自己特有的等待隊列傳給內核，讓內核把當前的進程掛載到其中）（遍歷數組中所有 fd）；
• 4.當設備就緒時，設備就會喚醒在自己特有等待隊列中的【所有】節點，于是當前進程就獲取到了完成的信號。poll文件操作返回的是一組標準的掩碼，其中的各個位指示當前的不同的就緒狀態（全0為沒有任何事件觸發），根據mask可對fd_set賦值；
• 5.如果所有設備返回的掩碼都沒有顯示任何的事件觸發，就去掉回調函數的函數指針，進入有限時的睡眠狀態，再恢復和不斷做poll，再作有限時的睡眠，直到其中一個設備有事件觸發為止。
• 6.只要有事件觸發，系統調用返回，將fd_set從內核空間拷貝到用戶空間，回到用戶態，用戶就可以對相關的fd作進一步的讀或者寫操作了。

3.select 優點

• select目前幾乎在所有的平臺上支持，良好跨平臺性。
• 單進程實現監視多個文件描述符，節省系統開銷

4.select 缺點

• 每次調用select，都需要把fd集合從用戶態拷貝到內核態，這個開銷在fd很多的時候會很大
• 單個進程能夠監視的fd數量存在最大限制，在linux上默認為1024（可以通過修改宏定義或者重新編譯內核的方式提升這個限制）
• 并且由于select的fd是放在數組中，并且每次都要線性遍歷整個數組，當fd很多的時候，開銷也很大

5.python select

調用select的函數為：

readable,writable,exceptional = select.select(rlist, wlist, xlist[, timeout])

前三個參數都分別是三個列表，數組中的對象均為waitable object:均是整數的文件描述符（file descriptor）或者一個擁有返回文件描述符方法fileno()的對象；

• rlist: 等待讀就緒的list
• wlist: 等待寫就緒的list
• errlist: 等待“異常”的list

select方法用來監視文件描述符，如果文件描述符發生變化，則獲取該描述符。

• 1、這三個list可以是一個空的list，但是接收3個空的list是依賴于系統的(在Linux上是可以接受的，但是在window上是不可以的)。
• 2、當 rlist 序列中的描述符發生可讀時（accetp和read），則獲取發生變化的描述符并添加到 readable 序列中
• 3、當 wlist 序列中含有描述符時，則將該序列中所有的描述符添加到 writable 序列中
• 4、當 errlist序列中的句柄發生錯誤時，則將該發生錯誤的句柄添加到 exceptional 序列中
• 5、當 超時時間 未設置，則select會一直阻塞，直到監聽的描述符發生變化當 超時時間 ＝ 1 時，那么如果監聽的句柄均無任何變化，則select會阻塞 1 秒，之后返回三個空列表，如果監聽的描述符（fd）有變化，則直接執行。
• 6、在list中可以接受Ptython的的file對象(比如sys.stdin，或者會被open()和os.open()返回的object)，socket object將會返回socket.socket()。也可以自定義類，只要有一個合適的fileno()的方法(需要真實返回一個文件描述符，而不是一個隨機的整數)。

6.select 示例:

#coding:UTF8import select import socket import sys import Queue#創建一個TCP/IP 進程 server = socket.socket(socket.AF_INET,socket.SOCK_STREAM) server.setblocking(0)#連接地址和端口 server_address = ('localhost',10000) print >>sys.stderr,'starting up on %s prot %s' % server_address server.bind(server_address)#最大允許鏈接數 server.listen(5)inputs = [ server ] outputs = []message_queues = {}while inputs:print >>sys.stderr,'\nwaiting for the next event'readable,writable,exceptional = select.select(inputs,outputs,inputs)# Handle inputsfor s in readable:if s is server:# A "readable" server socket is ready to accept a connectionconnection, client_address = s.accept()print >>sys.stderr, 'new connection from', client_address#connection.setblocking(0)inputs.append(connection)# Give the connection a queue for data we want to sendmessage_queues[connection] = Queue.Queue()else:data = s.recv(1024)if data:# A readable client socket has dataprint >>sys.stderr, 'received "%s" from %s' % (data, s.getpeername())message_queues[s].put(data) #這個s相當于connection# Add output channel for responseif s not in outputs:outputs.append(s)else:# Interpret empty result as closed connectionprint >>sys.stderr, 'closing', client_address, 'after reading no data'# Stop listening for input on the connectionif s in outputs:outputs.remove(s) #既然客戶端都斷開了，我就不用再給它返回數據了，所以這時候如果這個客戶端的連接對象還在outputs列表中，就把它刪掉inputs.remove(s) #inputs中也刪除掉s.close() #把這個連接關閉掉# Remove message queuedel message_queues[s]# Handle outputsfor s in writable:try:next_msg = message_queues[s].get_nowait()except Queue.Empty:# No messages waiting so stop checking for writability.print >>sys.stderr, 'output queue for', s.getpeername(), 'is empty'outputs.remove(s)else:print >>sys.stderr, 'sending "%s" to %s' % (next_msg, s.getpeername())s.send(next_msg.upper())# Handle "exceptional conditions"for s in exceptional:print >>sys.stderr, 'handling exceptional condition for', s.getpeername()# Stop listening for input on the connectioninputs.remove(s)if s in outputs:outputs.remove(s)s.close()# Remove message queuedel message_queues[s] import socket import queue from select import select SERVER_IP = ('127.0.0.1', 9999) # 保存客戶端發送過來的消息,將消息放入隊列中 message_queue = {} input_list = [] output_list = [] if __name__ == "__main__": server = socket.socket() server.bind(SERVER_IP) server.listen(10) # 設置為非阻塞 server.setblocking(False) # 初始化將服務端加入監聽列表 input_list.append(server) while True: # 開始 select 監聽,對input_list中的服務端server進行監聽 stdinput, stdoutput, stderr = select(input_list, output_list, input_list) # 循環判斷是否有客戶端連接進來,當有客戶端連接進來時select將觸發 for obj in stdinput: # 判斷當前觸發的是不是服務端對象, 當觸發的對象是服務端對象時,說明有新客戶端連接進來了 if obj == server: # 接收客戶端的連接, 獲取客戶端對象和客戶端地址信息 conn, addr = server.accept() print("Client {0} connected! ".format(addr)) # 將客戶端對象也加入到監聽的列表中, 當客戶端發送消息時 select 將觸發 input_list.append(conn) # 為連接的客戶端單獨創建一個消息隊列，用來保存客戶端發送的消息 message_queue[conn] = queue.Queue() else: # 由于客戶端連接進來時服務端接收客戶端連接請求，將客戶端加入到了監聽列表中(input_list)，客戶端發送消息將觸發 # 所以判斷是否是客戶端對象觸發 try: recv_data = obj.recv(1024) # 客戶端未斷開 if recv_data: print("received {0} from client {1}".format(recv_data.decode(), addr)) # 將收到的消息放入到各客戶端的消息隊列中 message_queue[obj].put(recv_data) # 將回復操作放到output列表中，讓select監聽 if obj not in output_list: output_list.append(obj) except ConnectionResetError: # 客戶端斷開連接了，將客戶端的監聽從input列表中移除 input_list.remove(obj) # 移除客戶端對象的消息隊列 del message_queue[obj] print("\n[input] Client {0} disconnected".format(addr)) # 如果現在沒有客戶端請求,也沒有客戶端發送消息時，開始對發送消息列表進行處理，是否需要發送消息 for sendobj in output_list: try: # 如果消息隊列中有消息,從消息隊列中獲取要發送的消息 if not message_queue[sendobj].empty(): # 從該客戶端對象的消息隊列中獲取要發送的消息 send_data = message_queue[sendobj].get() sendobj.sendall(send_data) else: # 將監聽移除等待下一次客戶端發送消息 output_list.remove(sendobj) except ConnectionResetError: # 客戶端連接斷開了 del message_queue[sendobj] output_list.remove(sendobj) print("\n[output] Client {0} disconnected".format(addr))

二、poll

1.相關概念

• poll在1986年誕生于System V Release3，它和select在本質上沒有多大差別，但是poll沒有最大文件描述符數量的限制。
• poll和select同樣存在一個缺點就是，包含大量文件描述符的數組被整體復制于用戶態和內核的地址空間之間，而不論這些文件描述符是否就緒，它的開銷隨著文件描述符數量的增加而線性增大。

另外，select()和poll()將就緒的文件描述符告訴進程后，如果進程沒有對其進行IO操作，那么下次調用select()和poll() 的時候將再次報告這些文件描述符，所以它們一般不會丟失就緒的消息，這種方式稱為水平觸發（Level Triggered）。

2.poll的原理

在Python中調用poll

select.poll()，返回一個poll的對象，支持注冊和注銷文件描述符。 poll.register(fd[, eventmask])注冊一個文件描述符

注冊后，可以通過poll()方法來檢查是否有對應的I/O事件發生。fd可以是i 個整數，或者有返回整數的fileno()方法對象。如果File對象實現了fileno()，也可以當作參數使用。

eventmask是一個你想去檢查的事件類型，它可以是常量POLLIN, POLLPRI和 POLLOUT的組合。如果缺省，默認會去檢查所有的3種事件類型。

事件常量意義

POLLIN	有數據讀取
POLLPRT	有數據緊急讀取
POLLOUT	準備輸出:輸出不會阻塞
POLLERR	某些錯誤情況出現
POLLHUP	掛起
POLLNVAL	無效請求:描述無法打開

poll.modify(fd, eventmask) 修改一個已經存在的fd，和poll.register (fd, eventmask)有相同的作用。如果去嘗試修改一個未經注冊的fd， 會引起一個errno為ENOENT的IOError。 poll.unregister(fd)從poll對象中注銷一個fd。嘗試去注銷一個未經注冊的fd， 會引起KeyError。 poll.poll([timeout])去檢測已經注冊了的文件描述符。會返回一個可能為空的 list，list中包含著(fd, event)這樣的二元組。fd是文件描述符， event是文件描述符對應的事件。如果返回的是一個空的list， 則說明超時了且沒有文件描述符有事件發生。timeout的單位是milliseconds， 如果設置了timeout，系統將會等待對應的時間。如果timeout缺省或者是None， 這個方法將會阻塞直到對應的poll對象有一個事件發生。

3.代碼

#coding: utf-8 import select, socketresponse = b"hello world"serversocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) serversocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) serversocket.bind(('localhost', 10000)) serversocket.listen(1) serversocket.setblocking(0)# poll = select.poll() poll.register(serversocket.fileno(), select.POLLIN)connections = {} while True:for fd, event in poll.poll():if event == select.POLLIN:if fd == serversocket.fileno():con, addr = serversocket.accept()poll.register(con.fileno(), select.POLLIN)connections[con.fileno()] = conelse:con = connections[fd]data = con.recv(1024)if data:poll.modify(con.fileno(), select.POLLOUT)elif event == select.POLLOUT:con = connections[fd]con.send(response)poll.unregister(con.fileno())con.close()

三、epoll

1.相關概念

直到Linux2.6才出現了由內核直接支持的實現方法，那就是epoll，它幾乎具備了之前所說的一切優點，被公認為Linux2.6下性能最好的多路I/O就緒通知方法。

epoll可以同時支持水平觸發和邊緣觸發（Edge Triggered，只告訴進程哪些文件描述符剛剛變為就緒狀態，它只說一遍，如果我們沒有采取行動，那么它將不會再次告知，這種方式稱為邊緣觸發），理論上邊緣觸發的性能要更高一些，但是代碼實現相當復雜。

• epoll同樣只告知那些就緒的文件描述符
• 當我們調用epoll_wait()獲得就緒文件描述符時，返回的不是實際的描述符，而是一個代表 就緒描述符數量的值
• 你只需要去epoll指定的一個數組中依次取得相應數量的文件描述符即可，這里也使用了內存映射（mmap）技術，這樣便徹底省掉了 這些文件描述符在系統調用時復制的開銷。
• 另一個本質的改進在于epoll采用基于事件的就緒通知方式。
• 在select/poll中，進程只有在調用一定的方法后，內核才對所有監視的文件描 述符進行掃描
• 而epoll事先通過epoll_ctl()來注冊一個文件描述符，一旦基于某個文件描述符就緒時，內核會采用類似callback的回調 機制，迅速激活這個文件描述符，當進程調用epoll_wait()時便得到通知。

2、epoll原理

調用epoll_create時，做了以下事情：

• 內核幫我們在epoll文件系統里建了個file結點；
• 在內核cache里建了個紅黑樹用于存儲以后epoll_ctl傳來的socket；
• 建立一個list鏈表，用于存儲準備就緒的事件。

調用epoll_ctl時，做了以下事情：

• 把socket放到epoll文件系統里file對象對應的紅黑樹上；
• 給內核中斷處理程序注冊一個回調函數，告訴內核，如果這個句柄的中斷到了，就把它放到準備就緒list鏈表里。

調用epoll_wait時，做了以下事情：

• 觀察list鏈表里有沒有數據。有數據就返回，沒有數據就sleep，等到timeout時間到后即使鏈表沒數據也返回。
• 通常情況下即使我們要監控百萬計的句柄，大多一次也只返回很少量的準備就緒句柄而已，所以，epoll_wait僅需要從內核態copy少量的句柄到用戶態而已。

3.優缺點

select

select本質上是通過設置或者檢查存放fd標志位的數據結構來進行下一步處理。這樣所帶來的缺點是(從聲明、到系統調用、到掃描、到返回后掃描)：

• 1).單個進程可監視的fd數量被限制
• 2).需要維護一個用來存放大量fd的數據結構，這樣會使得用戶空間和內核空間在傳遞該結構時復制開銷大
• 3).對socket進行掃描時是線性掃描
• 4).用戶也需要對返回的 fd_set 進行遍歷

poll

poll本質上和select沒有區別，它將用戶傳入的數組拷貝到內核空間，然后查詢每個fd對應的設備狀態，如果設備就緒則在設備等待隊列中加入一項并繼續遍歷，如果遍歷完所有fd后沒有發現就緒設備，則掛起當前進程，直到設備就緒或者主動超時，被喚醒后它又要再次遍歷fd。這個過程經歷了多次無謂的遍歷。

• poll沒有最大連接數的限制，原因是poll是基于數組來存儲的，但是同樣有一個缺點：
• 大量的fd的數組被整體復制于用戶態和內核地址空間之間，而不管這樣的復制是不是有意義。
• poll還有一個特點是“水平觸發”，如果報告了fd后，沒有被處理，那么下次poll時會再次報告該fd。

3.epoll

epoll支持水平觸發和邊緣觸發，最大的特點在于邊緣觸發，它只告訴進程哪些fd剛剛變為就需態，并且只會通知一次。在前面說到的復制問題上，epoll使用mmap減少復制開銷。還有一個特點是，epoll使用“事件”的就緒通知方式，通過epoll_ctl注冊fd，一旦該fd就緒，內核就會采用類似callback的回調機制來激活該fd，epoll_wait便可以收到通知

• 1.支持一個進程所能打開的最大連接數
• 2.select 單個進程所能打開的最大連接數有FD_SETSIZE宏定義，其大小是32個整數的大小（在32位的機器上，大小就是3232，同理64位機器上FD_SETSIZE為3264），當然我們可以對進行修改，然后重新編譯內核，但是性能可能會受到影響，這需要進一步的測試。
• 3.epoll本質上和select沒有區別，但是它沒有最大連接數的限制，原因是它是基于鏈表來存儲的
• 4.epoll 雖然連接數有上限，但是很大，1G內存的機器上可以打開10萬左右的連接，2G內存的機器可以打開20萬左右的連接

4. FD劇增后帶來的IO效率問題

• select 因為每次調用時都會對連接進行線性遍歷，所以隨著FD的增加會造成遍歷速度慢的“線性下降性能問題”。
• poll 同上
• epoll 因為epoll內核中實現是根據每個fd上的callback函數來實現的，只有活躍的socket才會主動調用callback，所以在活躍socket較少的情況下，使用epoll沒有前面兩者的線性下降的性能問題，但是所有socket都很活躍的情況下，可能會有性能問題。

5.消息傳遞方式

• select 內核需要將消息傳遞到用戶空間，都需要內核拷貝動作。
• poll 同上
• epoll epoll通過內核和用戶空間共享一塊內存來實現的。

下面我們對上面的socket例子進行改造，看一下select的例子：

4.在Python中調用epoll

select.epoll([sizehint=-1])返回一個epoll對象。 事件常量意義

EPOLLIN	讀就緒
EPOLLOUT	寫就緒
EPOLLPRI	有數據緊急讀取
EPOLLERR	assoc. fd有錯誤情況發生
EPOLLHUP	assoc. fd發生掛起
EPOLLRT	設置邊緣觸發(ET)（默認的是水平觸發）
EPOLLONESHOT	設置為 one-short 行為，一個事件(event)被拉出后，對應的fd在內部被禁用
EPOLLRDNORM	和 EPOLLIN 相等
EPOLLRDBAND	優先讀取的數據帶(data band)
EPOLLWRNORM	和 EPOLLOUT 相等
EPOLLWRBAND	優先寫的數據帶(data band)
EPOLLMSG 忽視

方法作用

epoll.close()	關閉epoll對象的文件描述符。
epoll.fileno	返回control fd的文件描述符number。
epoll.fromfd(fd)	用給予的fd來創建一個epoll對象。
epoll.register(fd[, eventmask])	在epoll對象中注冊一個文件描述符。（如果文件描述符已經存在，將會引起一個IOError）
epoll.modify(fd, eventmask)	修改一個已經注冊的文件描述符。
epoll.unregister(fd)	注銷一個文件描述符。
epoll.poll(timeout=-1[, maxevnets=-1])	等待事件，timeout(float)的單位是秒（second）。

5.例

#coding:Utf8 import socket, selectEOL1 = b'\n\n' EOL2 = b'\n\r\n' response = b'HTTP/1.0 200 OK\r\nDate: Mon, 1 Jan 1996 01:01:01 GMT\r\n' response += b'Content-Type: text/plain\r\nContent-Length: 13\r\n\r\n' response += b'Hello, world!'serversocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) serversocket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) serversocket.bind(('localhost', 10000)) serversocket.listen(1) serversocket.setblocking(0)epoll = select.epoll() epoll.register(serversocket.fileno(), select.EPOLLIN)try:connections = {}; requests = {}; responses = {}while True:events = epoll.poll(1)for fileno, event in events:if fileno == serversocket.fileno():connection, address = serversocket.accept()connection.setblocking(0)epoll.register(connection.fileno(), select.EPOLLIN)connections[connection.fileno()] = connectionrequests[connection.fileno()] = b''responses[connection.fileno()] = responseelif event & select.EPOLLIN:requests[fileno] += connections[fileno].recv(1024)if EOL1 in requests[fileno] or EOL2 in requests[fileno]:epoll.modify(fileno, select.EPOLLOUT)print('-'*40 + '\n' + requests[fileno].decode()[:-2])elif event & select.EPOLLOUT:byteswritten = connections[fileno].send(responses[fileno])responses[fileno] = responses[fileno][byteswritten:]if len(responses[fileno]) == 0:epoll.modify(fileno, 0)connections[fileno].shutdown(socket.SHUT_RDWR)elif event & select.EPOLLHUP:epoll.unregister(fileno)connections[fileno].close()del connections[fileno] finally:epoll.unregister(serversocket.fileno())epoll.close()serversocket.close() #!/usr/bin/env python import select import socket response = b'' serversocket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) serversocket.bind(('0.0.0.0', 8080)) serversocket.listen(1) # 因為socket默認是阻塞的，所以需要使用非阻塞（異步）模式。 serversocket.setblocking(0) # 創建一個epoll對象 epoll = select.epoll() # 在服務端socket上面注冊對讀event的關注。一個讀event隨時會觸發服務端socket去接收一個socket連接 epoll.register(serversocket.fileno(), select.EPOLLIN) try: # 字典connections映射文件描述符（整數）到其相應的網絡連接對象 connections = {} requests = {} responses = {} while True: # 查詢epoll對象，看是否有任何關注的event被觸發。參數“1”表示，我們會等待1秒來看是否有event發生。 # 如果有任何我們感興趣的event發生在這次查詢之前，這個查詢就會帶著這些event的列表立即返回 events = epoll.poll(1) # event作為一個序列（fileno，event code）的元組返回。fileno是文件描述符的代名詞，始終是一個整數。 for fileno, event in events: # 如果是服務端產生event,表示有一個新的連接進來 if fileno == serversocket.fileno(): connection, address = serversocket.accept() print('client connected:', address) # 設置新的socket為非阻塞模式 connection.setblocking(0) # 為新的socket注冊對讀（EPOLLIN）event的關注 epoll.register(connection.fileno(), select.EPOLLIN) connections[connection.fileno()] = connection # 初始化接收的數據 requests[connection.fileno()] = b'' # 如果發生一個讀event，就讀取從客戶端發送過來的新數據 elif event & select.EPOLLIN: print("------recvdata---------") # 接收客戶端發送過來的數據 requests[fileno] += connections[fileno].recv(1024) # 如果客戶端退出,關閉客戶端連接，取消所有的讀和寫監聽 if not requests[fileno]: connections[fileno].close() # 刪除connections字典中的監聽對象 del connections[fileno] # 刪除接收數據字典對應的句柄對象 del requests[connections[fileno]] print(connections, requests) epoll.modify(fileno, 0) else: # 一旦完成請求已收到，就注銷對讀event的關注，注冊對寫（EPOLLOUT）event的關注。寫event發生的時候，會回復數據給客戶端 epoll.modify(fileno, select.EPOLLOUT) # 打印完整的請求，證明雖然與客戶端的通信是交錯進行的，但數據可以作為一個整體來組裝和處理 print('-' * 40 + '\n' + requests[fileno].decode()) # 如果一個寫event在一個客戶端socket上面發生，它會接受新的數據以便發送到客戶端 elif event & select.EPOLLOUT: print("-------send data---------") # 每次發送一部分響應數據，直到完整的響應數據都已經發送給操作系統等待傳輸給客戶端 byteswritten = connections[fileno].send(requests[fileno]) requests[fileno] = requests[fileno][byteswritten:] if len(requests[fileno]) == 0: # 一旦完整的響應數據發送完成，就不再關注寫event epoll.modify(fileno, select.EPOLLIN) # HUP（掛起）event表明客戶端socket已經斷開（即關閉），所以服務端也需要關閉。 # 沒有必要注冊對HUP event的關注。在socket上面，它們總是會被epoll對象注冊 elif event & select.EPOLLHUP: print("end hup------") # 注銷對此socket連接的關注 epoll.unregister(fileno) # 關閉socket連接 connections[fileno].close() del connections[fileno] finally: # 打開的socket連接不需要關閉，因為Python會在程序結束的時候關閉。這里顯式關閉是一個好的代碼習慣 epoll.unregister(serversocket.fileno()) epoll.close() serversocket.close()

四、web靜態服務器-epool

以下代碼，支持http的長連接，即使用了Content-Length

import socket import time import sys import re import selectclass WSGIServer(object):"""定義一個WSGI服務器的類"""def __init__(self, port, documents_root):# 1. 創建套接字self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)# 2. 綁定本地信息self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)self.server_socket.bind(("", port))# 3. 變為監聽套接字self.server_socket.listen(128)self.documents_root = documents_root# 創建epoll對象self.epoll = select.epoll()# 將tcp服務器套接字加入到epoll中進行監聽self.epoll.register(self.server_socket.fileno(), select.EPOLLIN|select.EPOLLET)# 創建添加的fd對應的套接字self.fd_socket = dict()def run_forever(self):"""運行服務器"""# 等待對方鏈接while True:# epoll 進行 fd 掃描的地方 -- 未指定超時時間則為阻塞等待epoll_list = self.epoll.poll()# 對事件進行判斷for fd, event in epoll_list:# 如果是服務器套接字可以收數據，那么意味著可以進行acceptif fd == self.server_socket.fileno():new_socket, new_addr = self.server_socket.accept()# 向 epoll 中注冊 連接 socket 的 可讀 事件self.epoll.register(new_socket.fileno(), select.EPOLLIN | select.EPOLLET)# 記錄這個信息self.fd_socket[new_socket.fileno()] = new_socket# 接收到數據elif event == select.EPOLLIN:request = self.fd_socket[fd].recv(1024).decode("utf-8")if request:self.deal_with_request(request, self.fd_socket[fd])else:# 在epoll中注銷客戶端的信息self.epoll.unregister(fd)# 關閉客戶端的文件句柄self.fd_socket[fd].close()# 在字典中刪除與已關閉客戶端相關的信息del self.fd_socket[fd]def deal_with_request(self, request, client_socket):"""為這個瀏覽器服務器"""if not request:returnrequest_lines = request.splitlines()for i, line in enumerate(request_lines):print(i, line)# 提取請求的文件(index.html)# GET /a/b/c/d/e/index.html HTTP/1.1ret = re.match(r"([^/]*)([^ ]+)", request_lines[0])if ret:print("正則提取數據:", ret.group(1))print("正則提取數據:", ret.group(2))file_name = ret.group(2)if file_name == "/":file_name = "/index.html"# 讀取文件數據try:f = open(self.documents_root+file_name, "rb")except:response_body = "file not found, 請輸入正確的url"response_header = "HTTP/1.1 404 not found\r\n"response_header += "Content-Type: text/html; charset=utf-8\r\n"response_header += "Content-Length: %d\r\n" % len(response_body)response_header += "\r\n"# 將header返回給瀏覽器client_socket.send(response_header.encode('utf-8'))# 將body返回給瀏覽器client_socket.send(response_body.encode("utf-8"))else:content = f.read()f.close()response_body = contentresponse_header = "HTTP/1.1 200 OK\r\n"response_header += "Content-Length: %d\r\n" % len(response_body)response_header += "\r\n"# 將數據返回給瀏覽器client_socket.send(response_header.encode("utf-8")+response_body)# 設置服務器服務靜態資源時的路徑 DOCUMENTS_ROOT = "./html"def main():"""控制web服務器整體"""# python3 xxxx.py 7890if len(sys.argv) == 2:port = sys.argv[1]if port.isdigit():port = int(port)else:print("運行方式如: python3 xxx.py 7890")returnprint("http服務器使用的port:%s" % port)http_server = WSGIServer(port, DOCUMENTS_ROOT)http_server.run_forever()if __name__ == "__main__":main()

總結

以上是生活随笔為你收集整理的Python中的select、epoll详解的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。