1.epoll的由来
由于select只能同时处理1024个客户端,不能处理更多的客户端,于是linux系统就提供了epoll这种多路复用的IO模型,
注意:其他平台没有实现epoll模块,所以只能在linux中使用
2.程序阻塞过程分析
假设系统目前运行了三个进程 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对象的等待队列中
3.进程的唤醒
1.当网卡收到数据后会将数据入到缓冲区
2.发送中断信号给CPU
3.CPU执行中断程序,将数据从内核copy到socket的缓存区
4.唤醒进程,即将进程A切换到就绪状态,同时从socket的等待队列中移除这个进程引用
4.select监控多个socket
select的实现思路比较直接
1.先将所有socket放到列表中,
2.遍历这个列表,将进程A添加到每个socket的等待队列中,然后阻塞进程
3.当数据到达时,CPU执行中断程序将数据copy给socket,同时唤醒处于等待队列中的进程A,为了防止重复添加等待队列,
还需要移除已存在的进程A
4.进程A唤醒后,由于不清除那个socket有数据,所以需要遍历一遍所有cokset列表
从上面的过程中不难看出
1.select需要遍历socket列表,频繁的对等待队列进行添加移除操作
2.数据到达后还需要给遍历所有socket才能获取那些socket有数据,两个操作损耗的时间随着监控的socket的数量增大而增加,
处于效率才规定最大只能监视1024个socket
5.epoll要解决的问题
1.避免频繁的对等待队列进行操作
2避免遍历所有的socket
对于第一个问题我们先看select的处理方式
while True: r_list,w_list,x_list = select.select(rlist,wlist,xlist)
每次处理完一次读写后,都要将所用过重复一遍,包括移除进程,添加进程,默认就会将进程添加到等待队列,并阻塞住进程
然而等待队列的更新操作并不频繁
所以对于第一个问题epoll,采取的方案是将等待队列的维护和阻塞这两个操作进行拆分
相关待代码如下
import socket,selectserver = socket.socket()server.bind(("127.0.0.1",1688))server.listen(5)#创建epoll事件对象，后续要监控的事件添加到其中epoll = select.epoll()#注册服务器监听fd到等待读事件集合epoll.register(server.fileno(), select.EPOLLIN) # 需要关注 server这个socket的可读事件# 等待事件发生while True: for sock,event in epoll.poll(): pass 在epoll中register与unregister函数用于维护等待队列
register是进程添加到等待队列中unregister把进程从等待队列中删除
使用这两个函数我们自己来控制等待队列的添加删除从而避免频繁操作等待队列
epoll.poll则用于阻塞进程
这样一来就避免了每次处理都需要重新操作等待队列的问题
第二个问题是select中进程无法获取那些coket是有数据的所以需要遍历
eopll为了解这个问题,在内核中维护一个就绪列表
1.创建eopll对象,eopll也会对应一个文件,用文件系统管理
2.执行register,将eopll对象添加到socket的等待队列中
3.数据倒达后,CPU执行中断程序,将数据copy给socket
4.在epoll中,中断程序接下来回执行epoll对象中的回调函数,传入就绪的socket对象
5.将socket添加到就绪列表中
6.唤醒epoll等待队列中的进程,进程唤醒后由于存在就绪列表所以不需要再遍历socket了,直接处理列表即可
解决这两个问题后,并发量得到大幅度提升,最大可同时维护上万级的socket
6.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 socketclientsocket = 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)#关闭客户端socketclientsocket.close() 服务器
# coding:utf-8import socket, selectserver = 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)