完全可以自己写事件循环。 事件循环本质上就是一段用户态代码,不是操作系统的一部分,你完全可以用 Python 从零实现一个。


📌 事件循环的本质

事件循环的核心就是一个 while True 循环,不断做三件事:

  1. 检查有没有到期的定时任务
  2. 检查有没有就绪的 IO 事件(通过 epoll/select 等系统调用)
  3. 执行对应的回调函数
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 多路复用能力,事件循环本身是你(或库)在用户态写的调度代码。


Logo

openEuler 是由开放原子开源基金会孵化的全场景开源操作系统项目,面向数字基础设施四大核心场景(服务器、云计算、边缘计算、嵌入式),全面支持 ARM、x86、RISC-V、loongArch、PowerPC、SW-64 等多样性计算架构

更多推荐