LT（level trigger，水平触发）模式：当epoll_wait检测到描述符就绪，将此事件通知应用程序，应用程序可以不立即处理该事件。下次调用epoll_wait时，会再次响应应用程序并通知此事件。LT模式是默认的工作模式。
LT模式同时支持阻塞和非阻塞socket。
ET（edge trigger，边缘触发）模式：当epoll_wait检测到描述符就绪，将此事件通知应用程序，应用程序必须立即处理该事件。如果不处理，下次调用epoll_wait时，不会再次响应应用程序并通知此事件。
ET是高速工作方式，只支持非阻塞socket（ET模式减少了epoll事件被重复触发的次数，因此效率要比LT模式高）
Code提炼一下：
实例化对象：epoll = select.epoll()
注册对象：epoll.register(tcp_server.fileno(), select.EPOLLIN | select.EPOLLET)
注销对象：epoll.unregister(fd)
PS：epoll不一定比Select性能高，一般都是分场景的：
高并发下，连接活跃度不高时：epoll比Select性能高（eg：web请求，页面随时关闭）
并发不高，连接活跃度比较高：Select更合适（eg：小游戏）
Select是win和linux通用的，而epoll只有linux有
其实IO多路复用还有一个kqueue，和epoll类似，下面的通用写法中有包含
3.通用写法（Selector）一般来说：Linux下使用epoll，Win下使用select（IO多路复用会这个通用的即可）
先看看Python源代码：
# 选择级别：epoll|kqueue|devpoll > poll > select if 'KqueueSelector' in globals(): DefaultSelector = KqueueSelector elif 'EpollSelector' in globals(): DefaultSelector = EpollSelector elif 'DevpollSelector' in globals(): DefaultSelector = DevpollSelector elif 'PollSelector' in globals(): DefaultSelector = PollSelector else: DefaultSelector = SelectSelector实战案例：(可读和可写可以不分开)
import socket import selectors # Linux下使用epoll，Win下使用select Selector = selectors.DefaultSelector() class Task(object): def __init__(self): # 存放客户端fd和socket键值对 self.fd_socket_dict = dict() def run(self): self.server = socket.socket() self.server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) self.server.bind(('', 8080)) self.server.listen() # 把Server注册到epoll Selector.register(self.server.fileno(), selectors.EVENT_READ, self.connected) def connected(self, key): """客户端连接时处理""" client_socket, client_address = self.server.accept() fd = client_socket.fileno() self.fd_socket_dict[fd] = (client_socket, client_address) # 注册一个客户端读的事件（服务端去读消息） Selector.register(fd, selectors.EVENT_READ, self.call_back_reads) print(f"{client_address}已连接，当前连接数：{len(self.fd_socket_dict)}") def call_back_reads(self, key): """客户端可读时处理""" # 一个fd只能注册一次，监测可写的时候需要把可读给注销 Selector.unregister(key.fd) client_socket, client_address = self.fd_socket_dict[key.fd] print(f"[来自{client_address}的消息:]\n") data = client_socket.recv(2048) if data: print(data.decode("utf-8")) # 注册一个客户端写的事件（服务端去发消息） Selector.register(key.fd, selectors.EVENT_WRITE, self.call_back_writes) else: client_socket.close() del self.fd_socket_dict[key.fd] print(f"{client_address}已断开，当前连接数：{len(self.fd_socket_dict)}") def call_back_writes(self, key): """客户端可写时处理""" Selector.unregister(key.fd) client_socket, client_address = self.fd_socket_dict[key.fd] client_socket.send(b"ok") Selector.register(key.fd, selectors.EVENT_READ, self.call_back_reads) def main(): t = Task() t.run() while True: ready = Selector.select() for key, obj in ready: # 需要自己回调 call_back = key.data call_back(key) if __name__ == "__main__": main()Code提炼一下：
实例化对象：Selector = selectors.DefaultSelector()
注册对象：
Selector.register(server.fileno(), selectors.EVENT_READ, call_back)
Selector.register(server.fileno(), selectors.EVENT_WRITE, call_back)
注销对象：Selector.unregister(key.fd)
注意一下：一个fd只能注册一次，监测可写的时候需要把可读给注销（反之一样）
业余拓展：
select, iocp, epoll,kqueue及各种I/O复用机制 https://blog.csdn.net/shallwake/article/details/5265287 kqueue用法简介下级预估：协程篇 or 网络深入篇