传智播客 Web静态服务器-6-epoll

IO 多路复用

就是我们说的select，poll，epoll，有些地方也称这种IO方式为event driven IO。

select/epoll的好处就在于单个process就可以同时处理多个网络连接的IO。

它的基本原理就是select，poll，epoll这个function会不断的轮询所负责的所有socket，当某个socket有数据到达了，就通知用户进程。

所有的客户端遍历叫轮询，----事件通知效率高

核心是共享内存和事件通知

epoll简单模型

import socket
import select# 创建套接字
s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)# 设置可以重复使用绑定的信息
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR,1)# 绑定本机信息
s.bind(("",7788))# 变为被动
s.listen(10)# 创建一个epoll对象
epoll = select.epoll()# 测试，用来打印套接字对应的文件描述符
# print(s.fileno())
# print(select.EPOLLIN|select.EPOLLET)# 注册事件到epoll中
# epoll.register(fd[, eventmask])
# 注意，如果fd已经注册过，则会发生异常
# 将创建的套接字添加到epoll的事件监听中
epoll.register(s.fileno(), select.EPOLLIN|select.EPOLLET)connections = {}
addresses = {}# 循环等待客户端的到来或者对方发送数据
while True:# epoll 进行 fd 扫描的地方 -- 未指定超时时间则为阻塞等待epoll_list = epoll.poll()# 对事件进行判断for fd, events in epoll_list:# print fd# print events# 如果是socket创建的套接字被激活if fd == s.fileno():new_socket, new_addr = s.accept()print('有新的客户端到来%s' % str(new_addr))# 将 conn 和 addr 信息分别保存起来connections[new_socket.fileno()] = new_socketaddresses[new_socket.fileno()] = new_addr# 向 epoll 中注册 新socket 的 可读 事件epoll.register(new_socket.fileno(), select.EPOLLIN|select.EPOLLET)# 如果是客户端发送数据elif events == select.EPOLLIN:# 从激活 fd 上接收recvData = connections[fd].recv(1024).decode("utf-8")if recvData:print('recv:%s' % recvData)else:# 从 epoll 中移除该 连接 fdepoll.unregister(fd)# server 侧主动关闭该 连接 fdconnections[fd].close()print("%s---offline---" % str(addresses[fd]))del connections[fd]del addresses[fd]

说明

EPOLLIN （可读）
EPOLLOUT （可写）
EPOLLET （ET模式）

epoll对文件描述符的操作有两种模式：LT（level trigger）和ET（edge trigger）。LT模式是默认模式，LT模式与ET模式的区别如下：

LT模式：当epoll检测到描述符事件发生并将此事件通知应用程序，应用程序可以不立即处理该事件。下次调用epoll时，会再次响应应用程序并通知此事件。ET模式：当epoll检测到描述符事件发生并将此事件通知应用程序，应用程序必须立即处理该事件。如果不处理，下次调用epoll时，不会再次响应应用程序并通知此事件。

Epoll实现http linux

import socket
import re
import selectdef service_client(new_socket, request):"""为这个客户端返回数据"""# 1. 接收浏览器发送过来的请求 ，即http请求  # GET / HTTP/1.1# .....# request = new_socket.recv(1024).decode("utf-8")# print(">>>"*50)# print(request)request_lines = request.splitlines()print("")print(">"*20)print(request_lines)# GET /index.html HTTP/1.1# get post put delfile_name = ""ret = re.match(r"[^/]+(/[^ ]*)", request_lines[0])if ret:file_name = ret.group(1)# print("*"*50, file_name)if file_name == "/":file_name = "/index.html"# 2. 返回http格式的数据，给浏览器try:f = open("./html" + file_name, "rb")except:response = "HTTP/1.1 404 NOT FOUND\r\n"response += "\r\n"response += "------file not found-----"new_socket.send(response.encode("utf-8"))else:html_content = f.read()f.close()response_body = html_contentresponse_header = "HTTP/1.1 200 OK\r\n"response_header += "Content-Length:%d\r\n" % len(response_body)response_header += "\r\n"response = response_header.encode("utf-8") + response_bodynew_socket.send(response)def main():"""用来完成整体的控制"""# 1. 创建套接字tcp_server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)tcp_server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)# 2. 绑定tcp_server_socket.bind(("", 7890))# 3. 变为监听套接字tcp_server_socket.listen(128)tcp_server_socket.setblocking(False)  # 将套接字变为非堵塞# 创建一个epoll对象epl = select.epoll()# 将监听套接字对应的fd注册到epoll中epl.register(tcp_server_socket.fileno(), select.EPOLLIN)fd_event_dict = dict()while True:fd_event_list = epl.poll()  # 默认会堵塞，直到 os监测到数据到来 通过事件通知方式 告诉这个程序，此时才会解堵塞# [(fd, event), (套接字对应的文件描述符, 这个文件描述符到底是什么事件 例如 可以调用recv接收等)]for fd, event in fd_event_list:# 等待新客户端的链接if fd == tcp_server_socket.fileno():new_socket, client_addr = tcp_server_socket.accept()epl.register(new_socket.fileno(), select.EPOLLIN)fd_event_dict[new_socket.fileno()] = new_socketelif event==select.EPOLLIN:# 判断已经链接的客户端是否有数据发送过来recv_data = fd_event_dict[fd].recv(1024).decode("utf-8")if recv_data:service_client(fd_event_dict[fd], recv_data)else:fd_event_dict[fd].close()epl.unregister(fd)del fd_event_dict[fd]# 关闭监听套接字tcp_server_socket.close()if __name__ == "__main__":main()

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()

小总结

I/O 多路复用的特点：

通过一种机制使一个进程能同时等待多个文件描述符，而这些文件描述符（套接字描述符）其中的任意一个进入读就绪状态，epoll()函数就可以返回。所以, IO多路复用，本质上不会有并发的功能，因为任何时候还是只有一个进程或线程进行工作，它之所以能提高效率是因为select\epoll 把进来的socket放到他们的 '监视' 列表里面，当任何socket有可读可写数据立马处理，那如果select\epoll 手里同时检测着很多socket，一有动静马上返回给进程处理，总比一个一个socket过来,阻塞等待,处理高效率。

当然也可以多线程/多进程方式，一个连接过来开一个进程/线程处理，这样消耗的内存和进程切换页会耗掉更多的系统资源。所以我们可以结合IO多路复用和多进程/多线程来高性能并发，IO复用负责提高接受socket的通知效率，收到请求后，交给进程池/线程池来处理逻辑。

参考资料

如果想了解下epoll在Linux中的实现过程可以参考：http://blog.csdn.net/xiajun07061225/article/details/9250579