Cheney/Knowledge
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
Knowledge/0 编程/Python 下的 Select 模式和 EPoll 模式.md
T

11 KiB
Raw Permalink Blame History

一、select 模式相关函数

核心函数

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 模块常量

# 错误常量
select.error  # 异常类
select.EINTR  # 中断错误
select.EBADF  # 错误的文件描述符

# 跨平台常量
select.PIPE_BUF  # 管道缓冲区大小

示例代码

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）

核心类和方法

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 常量

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 边缘触发模式

示例代码

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 模式相关函数

核心类和方法

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 常量

select.POLLIN     # 1   可读
select.POLLPRI    # 2   紧急数据可读  
select.POLLOUT    # 4   可写
select.POLLERR    # 8   错误
select.POLLHUP    # 16  挂起
select.POLLNVAL   # 32  无效请求

示例代码

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) 事件通知
触发模式 水平触发 水平触发 水平/边缘触发
内存使用 固定大小位图 动态数组 红黑树+就绪链表

性能对比代码示例

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

通用封装示例

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. 文件描述符限制：

    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 View Git Blame Copy Permalink