並發編程-epoll模型的探索與實踐

前言

我們知道nginx的效率非常高,能處理上萬級的並發,其之所以高效離不開epoll的支持,

epoll是什么呢?,epoll是IO模型中的一種,屬於多路復用IO模型;

到這里你應該想到了,select,的確select也是一種多路復用的IO模型,但是其單個select最多只能同時處理1024個socket,效率實在算不上高,這時候epoll來救場了

本文從阻塞IO模型的基礎上展開討論,一步步靠近epoll的實現原理,最后以一個簡單的epoll案例程序作為結束

親手寫一個epoll,然后去虐面試官吧!

在select的學習過程中我們知道了select 只能同時處理1024個客戶端,

而多線程會遇到資源瓶頸,什么才是解決高並發最有效的方式呢

linux中提供了epoll 這種高效的多路復用IO模型

注意其他平台沒有相應的實現所以epoll僅在linux中可用

程序阻塞過程分析

epoll代碼實現並不復雜,但是要搞清楚其高效的原理還是需要花一些時間的

我們從最原始的阻塞模型開始分析

假設系統目前運行了三個進程 A B C

進程A正在運行一下socket程序

server = socket.socket()
server.bind(("127.0.0.1",1688))
server.listen()
server.accept()

1.系統會創建文件描述符指向一個socket對象 ,其包含了讀寫緩沖區,已經進行等待隊列

2.當執行到accept / recv 時系統會講進程A 從工作隊列中移除

3.將進程A的引用添加到 socket對象的等待隊列中

進程的喚醒

1.當網卡收到數據后會現將數據寫入到緩沖區

2.發送中斷信號給CPU

3.CPU執行中斷程序,將數據從內核copy到socket的緩沖區

4.喚醒進程,即將進程A切換到就緒態,同時從socket的等待隊列中移除這個進程引用

select監控多個socket

select的實現思路比較直接

1.先將所有socket放到一個列表中,

2.遍歷這個列表將進程A 添加到每個socket的等待隊列中 然后阻塞進程

3.當數據到達時,cpu執行中斷程序將數據copy給socket 同時喚醒處於等待隊列中的進程A

為了防止重復添加等待隊列 還需要移除已經存在的進程A

4.進程A喚醒后 由於不清楚那個socket有數據,所以需要遍歷一遍所有socket列表

從上面的過程中不難看出

1.select,需要遍歷socket列表,頻繁的對等待隊列進行添加移除操作,

2.數據到達后還需要給變量所有socket才能獲知哪些socket有數據

兩個操作消耗的時間隨着要監控的socket的數量增加而大大增加,

處於效率考慮才規定了最大只能監視1024個socket

epol要解決的問題

1.避免頻繁的對等待隊列進行操作

2.避免遍歷所有socket

對於第一個問題我們先看select的處理方式

while True:
	r_list,w_list,x_list = select.select(rlist,wlist,xlist)
  

每次處理完一次讀寫后,都需要將所有過沖重復一遍,包括移除進程,添加進程,默認就會將進程添加到等待隊列,並阻塞住進程,然而等待隊列的更新操作並不頻繁,

所以對於第一個問題epoll采取的方案是,將對等待隊列的維護和,阻塞進程這兩個操作進行拆分,

相關代碼如下

import socket,select
server = socket.socket()
server.bind(("127.0.0.1",1688))
server.listen(5)

#創建epoll事件對象，后續要監控的事件添加到其中
epoll = select.epoll()
#注冊服務器監聽fd到等待讀事件集合
epoll.register(server.fileno(), select.EPOLLIN)

# 等待事件發生
while True:
	for sock,event in epoll.poll():
    pass

在epoll中register 與 unregister函數用於維護等待隊列

epoll.poll則用於阻塞進程

這樣一來就避免了 每次處理都需要重新操作等待隊列的問題

第二個問題是select中進程無法獲知哪些socket是有數據的所以需要遍歷

epol為了解決這個問題,在內核中維護了一個就緒列表,

1.創建epoll對象,epoll也會對應一個文件,由文件系統管理

2.執行register時,將epoll對象 添加到socket的等待隊列中

3.數據到達后,CPU執行中斷程序,將數據copy給socket

4.在epoll中,中斷程序接下來會執行epoll對象中的回調函數,傳入就緒的socket對象

5.將socket,添加到就緒列表中

6.喚醒epoll等待隊列中的進程,

進程喚醒后,由於存在就緒列表,所以不需要再遍歷socket了,直接處理就緒列表即可

解決了這兩個問題后,並發量得到大幅度提升,最大可同時維護上萬級別的socket

epoll相關函數

import select 導入select模塊

epoll = select.epoll() 創建一個epoll對象

epoll.register(文件句柄,事件類型) 注冊要監控的文件句柄和事件

事件類型:

　　select.EPOLLIN    可讀事件

　　select.EPOLLOUT   可寫事件

　　select.EPOLLERR   錯誤事件

　　select.EPOLLHUP   客戶端斷開事件

epoll.unregister(文件句柄)   銷毀文件句柄

epoll.poll(timeout)  當文件句柄發生變化，則會以列表的形式主動報告給用戶進程,timeout

                     為超時時間，默認為-1，即一直等待直到文件句柄發生變化，如果指定為1

                     那么epoll每1秒匯報一次當前文件句柄的變化情況，如果無變化則返回空

epoll.fileno() 返回epoll的控制文件描述符(Return the epoll control file descriptor)

epoll.modfiy(fineno,event) fineno為文件描述符 event為事件類型  作用是修改文件描述符所對應的事件

epoll.fromfd(fileno) 從1個指定的文件描述符創建1個epoll對象

epoll.close()   關閉epoll對象的控制文件描述符

案例:

#coding:utf-8
#客戶端
#創建客戶端socket對象
import socket
clientsocket = socket.socket(socket.AF_INET,socket.SOCK_STREAM)
#服務端IP地址和端口號元組
server_address = ('127.0.0.1',1688)
#客戶端連接指定的IP地址和端口號
clientsocket.connect(server_address)

while True:
    #輸入數據
    data = raw_input('please input:')
    if data == "q":
        break
    if not data:
      continue
    #客戶端發送數據
    clientsocket.send(data.encode("utf-8"))
    #客戶端接收數據
    server_data = clientsocket.recv(1024)
    print ('客戶端收到的數據：',server_data)
#關閉客戶端socket
clientsocket.close()

服務器:

# coding:utf-8
import socket, select

server = socket.socket()
server.bind(("127.0.0.1", 1688))
server.listen(5)

msgs = []


fd_socket = {server.fileno(): server}
epoll = select.epoll()
# 注冊服務器的 寫就緒
epoll.register(server.fileno(), select.EPOLLIN)

while True:
    for fd, event in epoll.poll():
        sock = fd_socket[fd]
        print(fd, event)
        # 返回的是文件描述符 需要獲取對應socket
        if sock == server:  # 如果是服務器 就接受請求
            client, addr = server.accept()
            # 注冊客戶端寫就緒
            epoll.register(client.fileno(), select.EPOLLIN)
            # 添加對應關系
            fd_socket[client.fileno()] = client

        # 讀就緒
        elif event == select.EPOLLIN:
            data = sock.recv(2018)
            if not data:
                # 注銷事件
                epoll.unregister(fd)
                # 關閉socket
                sock.close()
                # 刪除socket對應關系
                del fd_socket[fd]
                print(" somebody fuck out...")
                continue

            print(data.decode("utf-8"))
            # 讀完數據 需要把數據發回去所以接下來更改為寫就緒=事件
            epoll.modify(fd, select.EPOLLOUT)
            #記錄數據
            msgs.append((sock,data.upper()))
        elif event == select.EPOLLOUT:
            for item in msgs[:]:
                if item[0] == sock:
                    sock.send(item[1])
                    msgs.remove(item)
            # 切換關注事件為寫就緒
            epoll.modify(fd,select.EPOLLIN)

注意:上述代碼只能在linux下運行,因為epoll模型是linux內核提供的,上層代碼無法實現!