利用epoll實現異步IO

　　之前異步IO一直沒搞明白，大致的理解就是在一個大的循環中，有兩部分：第一部分是監聽事件；第二部分是處理事件（通過添加回調函數的方式）。就拿網絡通信來說，可以先通過調用 select 模塊中的 select 監聽各個 socket 。當 socket 有事件到來時，針對相應的事件做出處理，就這么一直循環下去。所以異步IO也被稱為事件驅動IO。原理其實我說得太簡單了，所以我會以一個例子來說明一切。不過在這之前我還是要說一下 select 和 epoll 的區別。

一、IO多路服用的select

　　IO多路復用相對於阻塞式和非阻塞式的好處就是它可以監聽多個 socket ，並且不會消耗過多資源。當用戶進程調用 select 時，它會監聽其中所有 socket 直到有一個或多個 socket 數據已經准備好，否則就一直處於阻塞狀態。select的缺點在於單個進程能夠監視的文件描述符的數量存在最大限制，select()所維護的存儲大量文件描述符的數據結構，隨着文件描述符數量的增大，其復制的的開銷也線性增長。同時，由於網絡響應時間的延遲使得大量的tcp鏈接處於非常活躍狀態，但調用select()會對所有的socket進行一次線性掃描，所以這也浪費了一定的開銷。不過它的好處還有就是它的跨平台特性。

二、 異步IO的epoll

　　epoll的優點就是完全的異步，你只需要對其中 poll 函數注冊相應的 socket 和事件，就可以完全不管。當有時間發生時，數據已經從內核態拷貝到用戶態，也就是完全沒有阻塞。

 

三、基於epoll的聊天室程序

　　說了這么多，我決定還是用epoll寫一個多人聊天程序。epoll可以支持大量連接，select卻有限制，所以這就是我決定用epoll的原因。首先看服務器程序：

import socket, select
# 服務端

serverSock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
serverSock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, True)
serverSock.bind(('127.0.0.1', 8001))
serverSock.listen(5)
serverSock.setblocking(False)

EOL = bytes('\n\n', 'utf-8')
QUIT = bytes('\r\n', 'utf-8')
epoll = select.epoll()
epoll.register(serverSock.fileno(), select.EPOLLIN)
print('注冊事件：%s'%serverSock.fileno())

try:
    connections = {}; requests = {}; responses = {}
    while True:
        events = epoll.poll(1)
        for fileno, event in events:
#             print('event:%s fileno:%s'%(event, fileno))
            if fileno == serverSock.fileno():
                clientSock, address = serverSock.accept()
                print('連接到客戶端： %s：%s'%(address[0], address[1]))
                clientSock.setblocking(False)
                connections[clientSock.fileno()] = (clientSock, address)
                epoll.register(clientSock.fileno(), select.EPOLLOUT) # socket只能注冊輸入或輸出一個，不能同時注冊
                requests[clientSock.fileno()] = bytes('', 'utf-8')
                responses[clientSock.fileno()] = bytes('你已連接到服務器,IP為{}:{}\n\n'.format(*serverSock.getsockname()),
                                         'utf-8')
            elif event & select.EPOLLIN:
                requests[fileno] += connections[fileno][0].recv(1024)
                if requests[fileno].endswith(EOL):
                    msg = str(requests[fileno], 'utf-8')
                    msg = '來自{}的消息：{}'.format(connections[fileno][1], msg[:-2])
                    requests[fileno] = b''
                    #print(msg)
                    for i in responses:
                        if i == fileno:
                            continue
                        responses[i] += bytes(msg, 'utf-8')
                        epoll.modify(i, select.EPOLLOUT)
                if QUIT in requests[fileno]:
                    epoll.modify(fileno, select.EPOLLOUT) 
                    
            elif event & select.EPOLLOUT:
                #print('開始發送消息：%s'%str(responses[fileno], 'utf-8'))
                bytesSend = connections[fileno][0].send(responses[fileno])
                responses[fileno] = responses[fileno][bytesSend:]
                #print('發送完成')
                if responses[fileno] == b'':
                    epoll.modify(fileno, select.EPOLLIN)
                if QUIT in requests[fileno]:
                    epoll.modify(fileno, 0)
                    connections[fileno][0].shutdown(socket.SHUT_RDWR)
                    
            elif event & select.EPOLLHUP:
                epoll.unregister(fileno)
                connections[fileno][0].close()
                del connections[fileno]
finally:
    epoll.unregister(serverSock.fileno())
    epoll.close()
    serverSock.close()
    print('已退出服務端程序')

 

 

注意，我首先定義了兩個終止符：EOL表示這段話已經發完了；QUIT表示客戶端想要退出。客戶端的程序有點讓我為難，既要在命令行輸入又要同時保證能輸出別人發過來的消息，所有我只好用了prompt_toolkit再加上一個線程。如下：

import socket, prompt_toolkit, select
import threading, queue


class Client:
    def __init__(self, sock):
        self.sock = sock
        self.want_to_send = False
        self.want_to_recv = True
        self._msg = queue.Queue()
        
    def fileno(self):
        return self.sock.fileno()
    
    def handle_recv(self):
        print('接受消息..')
        msg = self.sock.recv(1024)
        print(str(msg, 'utf-8'))
        
    def handle_send(self):
        msg = self._msg.get()
        if msg == '\r\n':
            self.want_to_send = False
            self.want_to_recv = False
        self.sock.sendall(bytes(msg, 'utf-8'))
        self.want_to_send = False
            
def handle_sock(want_to_send, want_to_recv, sock):
    print('開始處理消息...')
    want_to_recv.append(sock.fileno())
    while True:
        if sock.want_to_send:
            if not want_to_send:
                want_to_send.append(myclient.fileno())
        else:
            want_to_send.clear()
        can_recv, can_send, _ = select.select(want_to_recv, want_to_send, [], 1)
        if can_recv:
            sock.handle_recv()
        if can_send:
            sock.handle_send()
        if not (sock.want_to_send or sock.want_to_recv):
            print('正停止客戶端連接...')
            break
        if sock._msg.qsize():
            sock.want_to_send = True

            
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
s.connect(('127.0.0.1',8001))

myclient = Client(s)
want_to_send = []
want_to_recv = []
    

            
t = threading.Thread(target=handle_sock,
                     args=(want_to_send, want_to_recv, myclient),
                     daemon=True)
t.start()

try:
    while True:
        messages = prompt_toolkit.shortcuts.prompt('\n\n>>> ',patch_stdout=True)
        myclient._msg.put(messages+'\n\n')
except KeyboardInterrupt:
    myclient._msg.put('\r\n')
finally:
    t.join()
    myclient.sock.close()
    print('網絡已斷開')

 

我的服務器跑在 jupyter 上，客戶端跑在命令行上，效果如下：

 

　　客戶端接受和發送消息都是互不影響的，這樣就實現了一個多人聊天的功能。而且服務器使用的是epoll，所以哪怕是成千上萬的人同時在線也沒有任何壓力。至於怎么測試暫時還沒想到辦法。