可以自己写一个异步的事件循环吗?
·
完全可以自己写事件循环。 事件循环本质上就是一段用户态代码,不是操作系统的一部分,你完全可以用 Python 从零实现一个。
📌 事件循环的本质
事件循环的核心就是一个 while True 循环,不断做三件事:
- 检查有没有到期的定时任务
- 检查有没有就绪的 IO 事件(通过 epoll/select 等系统调用)
- 执行对应的回调函数
while True:
检查定时器 → 到期的任务放入就绪队列
检查 IO 事件 → 有事件的回调放入就绪队列
执行就绪队列里的所有回调
就这么简单。Python 的 asyncio、Node.js 的 libuv、C++ 的 boost::asio,底层都是这个模式。
📌 极简版:自己写一个事件循环
下面是一个能跑的最小事件循环,支持立即调度和延迟调度:
import time
from collections import deque
class SimpleEventLoop:
def __init__(self):
self._ready = deque() # 就绪任务队列
self._scheduled = [] # 定时任务列表
self._stopping = False
def call_soon(self, callback, *args):
"""立即调度一个任务"""
self._ready.append((callback, args))
def call_later(self, delay, callback, *args):
"""延迟 delay 秒后调度一个任务"""
scheduled_time = time.time() + delay
self._scheduled.append((scheduled_time, callback, args))
self._scheduled.sort(key=lambda x: x[0]) # 按时间排序
def run_forever(self):
"""事件循环主体"""
while not self._stopping:
# 1. 检查定时任务,到期的放入就绪队列
now = time.time()
while self._scheduled and self._scheduled[0][0] <= now:
_, callback, args = self._scheduled.pop(0)
self._ready.append((callback, args))
# 2. 执行所有就绪任务
while self._ready:
callback, args = self._ready.popleft()
callback(*args)
# 3. 没有任务就小睡一下,避免 CPU 空转
if not self._ready:
time.sleep(0.01)
def stop(self):
self._stopping = True
使用它
loop = SimpleEventLoop()
def hello(name):
print(f"Hello, {name}!")
def stop_loop():
print("停止事件循环")
loop.stop()
# 立即执行
loop.call_soon(hello, "Alice")
# 延迟 2 秒执行
loop.call_later(2, hello, "Bob")
# 延迟 1 秒执行
loop.call_later(1, hello, "Charlie")
# 延迟 3 秒停止循环
loop.call_later(3, stop_loop)
print("开始运行自定义事件循环...")
loop.run_forever()
print("事件循环结束")
输出
开始运行自定义事件循环...
Hello, Alice!
Hello, Charlie!
Hello, Bob!
停止事件循环
事件循环结束
📌 进阶版:加上 IO 监听
上面的版本只支持定时任务,如果要监听网络 IO(比如 socket 可读),只需要加上 selectors 模块(底层就是 epoll):
import selectors
sel = selectors.DefaultSelector()
# 注册一个 socket 的可读事件
sel.register(sock, selectors.EVENT_READ, on_data_ready)
# 事件循环里加上:
events = sel.select(timeout=0.1) # 等待 IO 事件,最多等 0.1 秒
for key, mask in events:
callback = key.data
callback() # 调用对应的回调
这就是 Python asyncio 底层干的事情——用 selectors(epoll)监听 IO 事件,事件就绪后恢复对应的协程。
📌 现实中的"自定义事件循环"
其实你已经在用了:
| 事件循环 | 说明 |
|---|---|
| asyncio 默认循环 | Python 自带的,纯 Python 实现 |
| uvloop | 用 C 写的,基于 libuv(和 Node.js 同一个库),比默认快 2~4 倍 |
| Node.js 事件循环 | 基于 libuv 实现 |
| Qt 事件循环 | GUI 框架自己的事件循环 |
比如 FastAPI 用的 Uvicorn,默认就会尝试用 uvloop 替代 asyncio 的默认事件循环,因为性能更好。
📌 总结
事件循环 = 一个 while 循环 + 一个就绪队列 + 一个定时器 + 一个 IO 监听器。 操作系统只提供底层的 epoll/select 等 IO 多路复用能力,事件循环本身是你(或库)在用户态写的调度代码。
openEuler 是由开放原子开源基金会孵化的全场景开源操作系统项目,面向数字基础设施四大核心场景(服务器、云计算、边缘计算、嵌入式),全面支持 ARM、x86、RISC-V、loongArch、PowerPC、SW-64 等多样性计算架构
更多推荐

所有评论(0)