You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Python中select.select与asyncio.wait对比:该选用哪一种?

问题

select.select与asyncio.wait均可用于等待一组文件中首个变为可读的文件。已知select调用UNIX系统调用,asyncio.wait基于协程(绿色线程)实现,请问:

  1. 两种方案理论上各有哪些优缺点?
  2. 它们是否使用相同的底层原语?
  3. 具体而言,asyncio.wait在等待文件描述符时是否会调用select?

一、优缺点对比

select.select的优缺点

  • 优点:
    • 直接调用系统select,无额外封装开销,性能损耗极低
    • 同步阻塞模型,逻辑简单直观,无需理解协程/异步编程概念
    • 兼容性极强,几乎所有UNIX-like系统都支持,依赖少
  • 缺点:
    • 同步阻塞会占用整个线程,线程无法并行处理其他任务,资源利用率低
    • 单个select能监听的文件描述符数量受系统限制(默认通常是1024)
    • 每次调用需要遍历所有监听的描述符,描述符数量越多,效率越低

asyncio.wait的优缺点

  • 优点:
    • 基于异步协程模型,单线程即可处理大量并发任务,资源利用率高
    • 支持扩展复杂异步逻辑,比如同时处理IO、定时任务等场景
    • 不受select的文件描述符数量限制(底层可能采用更高效的多路复用机制)
    • 任务取消、超时控制等功能更完善
  • 缺点:
    • 需要适配异步API(比如示例中的aiofiles),有一定学习成本
    • 异步框架存在调度开销,简单场景下性能不如原生select
    • 异步逻辑调试比同步代码更复杂,问题排查难度大

二、底层原语是否相同?

两者不完全相同:

  • select.select直接封装了UNIX系统的select()系统调用,是最基础的IO多路复用原语之一。
  • asyncio的底层IO多路复用机制是可替换的:在不同系统或配置下,会自动选择最优的多路复用原语,包括select、poll、epoll(Linux)、kqueue(BSD/macOS)等。asyncio.wait本身是协程任务的调度接口,其底层依赖于asyncio事件循环的多路复用实现。

三、asyncio.wait是否会调用select?

有可能,但不是必然:

  • 当asyncio事件循环使用select作为多路复用后端时(比如某些旧系统或手动指定SelectorEventLoop),asyncio.wait在等待文件描述符就绪时,底层会调用系统select。
  • 现代Linux系统中,asyncio默认会使用epoll;在macOS/BSD则默认用kqueue,这些情况下不会调用select,而是使用更高效的系统原语。
  • 示例中使用aiofiles打开设备文件,其底层会将文件描述符注册到asyncio事件循环的多路复用器中,当调用asyncio.wait时,事件循环会通过对应的多路复用原语(可能是select/epoll/kqueue等)监听就绪状态。

代码示例

同步方案(select.select)

import select
devices = ["/dev/input/event6", "/dev/input/event3"]
files = [open(device) for device in devices]
first_fileno = select.select([file.fileno() for file in files], [], [])[0][0]
print([file.name for file in files if file.fileno() == first_fileno][0])

异步方案(asyncio.wait)

import aiofiles
import asyncio
import contextlib

async def read_and_return(file):
    contents = await file.read(1)
    return file, contents

async def main(devices):
    async with contextlib.AsyncExitStack() as stack:
        files = [await stack.enter_async_context(aiofiles.open(device, "rb")) for device in devices]
        tasks = [asyncio.create_task(read_and_return(file)) for file in files]
        completed, incompleted = await asyncio.wait(tasks, return_when=asyncio.FIRST_COMPLETED)
        file, contents = list(completed)[0].result()
        print(file.name)

devices = ["/dev/input/event6", "/dev/input/event3"]
asyncio.run(main(devices))

内容的提问来源于stack exchange,提问作者charmoniumQ

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.07.13 16:13:26