vault backup: 2026-11-07 16:33:44
This commit is contained in:
cheney
@@ -0,0 +1,390 @@
|
||||
## 一、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）
|
||||
Reference in New Issue
Block a user