390 lines
11 KiB
Markdown
390 lines
11 KiB
Markdown
## 一、select 模式相关函数
|
||
|
||
### 核心函数
|
||
```python
|
||
import select
|
||
|
||
# 1. 主要的 select 函数
|
||
select.select(rlist, wlist, xlist[, timeout])
|
||
# rlist: 等待读就绪的socket列表
|
||
# wlist: 等待写就绪的socket列表
|
||
# xlist: 等待异常条件的socket列表
|
||
# timeout: 超时时间（秒），None表示阻塞，0表示立即返回
|
||
|
||
# 2. 在 Windows 上使用
|
||
select.win32_select(rlist, wlist, xlist[, timeout])
|
||
```
|
||
|
||
### select 模块常量
|
||
```python
|
||
# 错误常量
|
||
select.error # 异常类
|
||
select.EINTR # 中断错误
|
||
select.EBADF # 错误的文件描述符
|
||
|
||
# 跨平台常量
|
||
select.PIPE_BUF # 管道缓冲区大小
|
||
```
|
||
|
||
### 示例代码
|
||
```python
|
||
import select
|
||
import socket
|
||
|
||
# 创建服务器
|
||
server = socket.socket()
|
||
server.bind(('localhost', 8888))
|
||
server.listen(5)
|
||
|
||
# 要监视的socket列表
|
||
read_list = [server]
|
||
write_list = []
|
||
error_list = []
|
||
|
||
while True:
|
||
# 调用 select
|
||
readable, writable, exceptional = select.select(
|
||
read_list, write_list, error_list, 1.0
|
||
)
|
||
|
||
for s in readable:
|
||
if s is server:
|
||
# 新连接
|
||
client, addr = server.accept()
|
||
read_list.append(client)
|
||
else:
|
||
# 可读数据
|
||
data = s.recv(1024)
|
||
if data:
|
||
# 处理数据
|
||
write_list.append(s) # 准备回写
|
||
else:
|
||
# 连接关闭
|
||
read_list.remove(s)
|
||
s.close()
|
||
|
||
for s in writable:
|
||
s.send(b"Response")
|
||
write_list.remove(s)
|
||
```
|
||
|
||
## 二、epoll 模式相关函数（仅限Linux）
|
||
|
||
### 核心类和方法
|
||
```python
|
||
import select
|
||
|
||
# 1. 创建 epoll 对象
|
||
epoll = select.epoll([sizehint=-1])
|
||
|
||
# 2. epoll 对象的方法
|
||
epoll.register(fd[, eventmask])
|
||
# fd: 文件描述符
|
||
# eventmask: 事件掩码，可以是以下组合：
|
||
# select.EPOLLIN - 可读
|
||
# select.EPOLLOUT - 可写
|
||
# select.EPOLLPRI - 紧急数据可读
|
||
# select.EPOLLERR - 错误
|
||
# select.EPOLLHUP - 挂起
|
||
# select.EPOLLET - 边缘触发模式
|
||
# select.EPOLLONESHOT - 一次性事件
|
||
|
||
epoll.unregister(fd) # 注销文件描述符
|
||
epoll.modify(fd, eventmask) # 修改事件掩码
|
||
|
||
epoll.poll([timeout=-1[, maxevents=-1]])
|
||
# 返回 [(fd, events), ...] 列表
|
||
# timeout: 超时秒数，-1表示阻塞，0表示立即返回
|
||
|
||
epoll.close() # 关闭 epoll 对象
|
||
epoll.fileno() # 获取 epoll 文件描述符
|
||
```
|
||
|
||
### epoll 常量
|
||
```python
|
||
select.EPOLLIN # 1 可读
|
||
select.EPOLLPRI # 2 紧急数据可读
|
||
select.EPOLLOUT # 4 可写
|
||
select.EPOLLERR # 8 错误
|
||
select.EPOLLHUP # 16 挂起
|
||
select.EPOLLRDNORM # 64 普通数据可读
|
||
select.EPOLLRDBAND # 128 优先带数据可读
|
||
select.EPOLLWRNORM # 256 普通数据可写
|
||
select.EPOLLWRBAND # 512 优先带数据可写
|
||
select.EPOLLMSG # 1024
|
||
select.EPOLLRDHUP # 8192 对端关闭连接或半关闭
|
||
select.EPOLLEXCLUSIVE # 1<<28 独占唤醒
|
||
select.EPOLLWAKEUP # 1<<29
|
||
select.EPOLLONESHOT # 1<<30 一次性事件
|
||
select.EPOLLET # 1<<31 边缘触发模式
|
||
```
|
||
|
||
### 示例代码
|
||
```python
|
||
import select
|
||
import socket
|
||
|
||
# 创建 epoll 对象
|
||
epoller = select.epoll()
|
||
|
||
# 创建服务器socket
|
||
server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
|
||
server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
|
||
server.bind(('0.0.0.0', 8888))
|
||
server.listen(128)
|
||
server.setblocking(False)
|
||
|
||
# 注册服务器socket，监听可读事件
|
||
epoller.register(server.fileno(), select.EPOLLIN)
|
||
|
||
# 文件描述符到socket的映射
|
||
fd_to_socket = {server.fileno(): server}
|
||
|
||
try:
|
||
while True:
|
||
# 等待事件，超时1秒
|
||
events = epoller.poll(1)
|
||
|
||
if not events:
|
||
continue # 超时，继续等待
|
||
|
||
for fd, event in events:
|
||
sock = fd_to_socket[fd]
|
||
|
||
# 服务器socket可读，表示有新连接
|
||
if sock is server:
|
||
client, addr = server.accept()
|
||
client.setblocking(False)
|
||
epoller.register(client.fileno(),
|
||
select.EPOLLIN | select.EPOLLET) # 边缘触发
|
||
fd_to_socket[client.fileno()] = client
|
||
print(f"New connection: {addr}")
|
||
|
||
# 可读事件
|
||
elif event & select.EPOLLIN:
|
||
data = sock.recv(1024)
|
||
if data:
|
||
print(f"Received: {data}")
|
||
# 修改为监听可写事件
|
||
epoller.modify(fd, select.EPOLLOUT)
|
||
else:
|
||
# 连接关闭
|
||
epoller.unregister(fd)
|
||
sock.close()
|
||
del fd_to_socket[fd]
|
||
|
||
# 可写事件
|
||
elif event & select.EPOLLOUT:
|
||
sock.send(b"HTTP/1.1 200 OK\r\n\r\nHello")
|
||
# 发送完后继续监听可读事件
|
||
epoller.modify(fd, select.EPOLLIN)
|
||
|
||
# 错误事件
|
||
elif event & select.EPOLLERR:
|
||
epoller.unregister(fd)
|
||
sock.close()
|
||
del fd_to_socket[fd]
|
||
|
||
finally:
|
||
epoller.unregister(server.fileno())
|
||
epoller.close()
|
||
server.close()
|
||
```
|
||
|
||
## 三、poll 模式相关函数
|
||
|
||
### 核心类和方法
|
||
```python
|
||
import select
|
||
|
||
# 1. 创建 poll 对象
|
||
poller = select.poll()
|
||
|
||
# 2. poll 对象的方法
|
||
poller.register(fd[, eventmask])
|
||
# eventmask 可以是以下组合：
|
||
# select.POLLIN - 可读
|
||
# select.POLLPRI - 紧急数据可读
|
||
# select.POLLOUT - 可写
|
||
# select.POLLERR - 错误
|
||
# select.POLLHUP - 挂起
|
||
# select.POLLNVAL - 无效请求
|
||
|
||
poller.unregister(fd) # 注销文件描述符
|
||
poller.modify(fd, eventmask) # 修改事件掩码
|
||
|
||
poller.poll([timeout])
|
||
# 返回 [(fd, events), ...] 列表
|
||
# timeout: 超时毫秒数，None表示阻塞，0表示立即返回，负值也表示阻塞
|
||
```
|
||
|
||
### poll 常量
|
||
```python
|
||
select.POLLIN # 1 可读
|
||
select.POLLPRI # 2 紧急数据可读
|
||
select.POLLOUT # 4 可写
|
||
select.POLLERR # 8 错误
|
||
select.POLLHUP # 16 挂起
|
||
select.POLLNVAL # 32 无效请求
|
||
```
|
||
|
||
### 示例代码
|
||
```python
|
||
import select
|
||
import socket
|
||
|
||
# 创建 poll 对象
|
||
poller = select.poll()
|
||
|
||
# 创建服务器socket
|
||
server = socket.socket()
|
||
server.bind(('localhost', 8888))
|
||
server.listen(5)
|
||
|
||
# 注册服务器socket
|
||
poller.register(server, select.POLLIN)
|
||
|
||
# socket到事件掩码的映射（用于跟踪）
|
||
socket_to_mask = {server: select.POLLIN}
|
||
|
||
try:
|
||
while True:
|
||
# 等待事件，超时1000毫秒
|
||
events = poller.poll(1000)
|
||
|
||
for fd_event in events:
|
||
sock = fd_event[0]
|
||
event = fd_event[1]
|
||
|
||
# 新连接
|
||
if sock is server:
|
||
if event & select.POLLIN:
|
||
client, addr = server.accept()
|
||
poller.register(client, select.POLLIN)
|
||
socket_to_mask[client] = select.POLLIN
|
||
|
||
# 可读事件
|
||
elif event & select.POLLIN:
|
||
data = sock.recv(1024)
|
||
if data:
|
||
print(f"Received: {data}")
|
||
# 修改为监听可写事件
|
||
poller.modify(sock, select.POLLOUT)
|
||
socket_to_mask[sock] = select.POLLOUT
|
||
else:
|
||
# 连接关闭
|
||
poller.unregister(sock)
|
||
sock.close()
|
||
del socket_to_mask[sock]
|
||
|
||
# 可写事件
|
||
elif event & select.POLLOUT:
|
||
sock.send(b"Response")
|
||
# 改回监听可读事件
|
||
poller.modify(sock, select.POLLIN)
|
||
socket_to_mask[sock] = select.POLLIN
|
||
|
||
finally:
|
||
poller.unregister(server)
|
||
server.close()
|
||
```
|
||
|
||
## 四、选择指南
|
||
|
||
| 特性 | select | poll | epoll |
|
||
|------|--------|------|-------|
|
||
| **跨平台** | ✓ 所有平台 | ✗ 仅Unix | ✗ 仅Linux |
|
||
| **最大描述符数** | 通常1024 | 无限制 | 无限制 |
|
||
| **效率** | O(n) 每次遍历 | O(n) 每次遍历 | O(1) 事件通知 |
|
||
| **触发模式** | 水平触发 | 水平触发 | 水平/边缘触发 |
|
||
| **内存使用** | 固定大小位图 | 动态数组 | 红黑树+就绪链表 |
|
||
|
||
### 性能对比代码示例
|
||
```python
|
||
import select
|
||
import time
|
||
|
||
def benchmark(io_method='select', num_fds=100):
|
||
"""测试不同I/O多路复用方法的性能"""
|
||
|
||
if io_method == 'select':
|
||
# select 测试
|
||
rlist = list(range(num_fds))
|
||
start = time.perf_counter()
|
||
for _ in range(1000):
|
||
select.select([], rlist, [], 0.001)
|
||
return time.perf_counter() - start
|
||
|
||
elif io_method == 'poll':
|
||
# poll 测试
|
||
poller = select.poll()
|
||
start = time.perf_counter()
|
||
for _ in range(1000):
|
||
poller.poll(1) # 1ms timeout
|
||
return time.perf_counter() - start
|
||
```
|
||
|
||
### 通用封装示例
|
||
```python
|
||
import select
|
||
import sys
|
||
|
||
class Multiplexer:
|
||
"""多路复用器的通用封装"""
|
||
|
||
def __init__(self, backend='auto'):
|
||
if backend == 'auto':
|
||
if hasattr(select, 'epoll'):
|
||
self.backend = 'epoll'
|
||
self._impl = select.epoll()
|
||
elif hasattr(select, 'poll'):
|
||
self.backend = 'poll'
|
||
self._impl = select.poll()
|
||
else:
|
||
self.backend = 'select'
|
||
self._impl = None
|
||
else:
|
||
self.backend = backend
|
||
if backend == 'epoll':
|
||
self._impl = select.epoll()
|
||
elif backend == 'poll':
|
||
self._impl = select.poll()
|
||
else:
|
||
self._impl = None
|
||
|
||
def register(self, sock, events):
|
||
"""注册socket"""
|
||
if self.backend == 'select':
|
||
# select需要手动管理列表
|
||
pass
|
||
elif self.backend == 'poll':
|
||
self._impl.register(sock, events)
|
||
elif self.backend == 'epoll':
|
||
self._impl.register(sock.fileno(), events)
|
||
```
|
||
|
||
## 五、注意事项
|
||
|
||
1. **平台兼容性**：
|
||
- Windows只支持select
|
||
- macOS/BSD支持select和kqueue（通过select.kqueue）
|
||
- Linux支持所有三种
|
||
|
||
2. **边缘触发(ET) vs 水平触发(LT)**：
|
||
- epoll支持两种模式
|
||
- ET模式需要一次读取所有数据
|
||
- LT模式是默认的，select和poll只支持LT
|
||
|
||
3. **文件描述符限制**：
|
||
```python
|
||
import resource
|
||
|
||
# 查看和设置文件描述符限制
|
||
soft, hard = resource.getrlimit(resource.RLIMIT_NOFILE)
|
||
resource.setrlimit(resource.RLIMIT_NOFILE, (hard, hard))
|
||
```
|
||
|
||
选择建议：
|
||
- 小规模连接（<1000）：使用select或poll
|
||
- Linux高并发：使用epoll边缘触发
|
||
- 跨平台需求：使用select或第三方库（如asyncio） |