当前位置：首页 > news >正文

Python异步编程asyncio （一）：Event Loop

news 2026/3/27 9:04:46

1 Event loop

1.1 什么是Event Loop

Event Loop 是每个 asyncio 应用程序的核心。它在循环中运行，执行任务、处理事件并管理异步操作。

1.2 如何运行 Event Loop

我们使用 asyncio.get_event_loop() 获取 Event Loop 的引用：

import asyncio
loop = asyncio.get_event_loop()

我们可以使用 Event Loop 来运行异步任务。有两种常见的方法来运行：

import asyncio
loop = asyncio.get_event_loop()# 方法 1：run_forever()
loop.run_forever()# 方法 2：run_until_complete()
loop.run_until_complete(asyncio.sleep(1))

根据 python 中的 docstring 定义， run_forever() 的功能是 Run the event loop until stop() is called. ，而 run_until_complete() 的功能是 Run the event loop until a Future is done.，也就是 asyncio.sleep(1)）完成。

1.3 EventLoop 就是一个 While 循环

在 Python 3.14 源码的 Lib\asyncio\base_events.py 内， run_forever 方法的实现非常直白：


def run_forever(self):
    """Run until stop() is called."""
    self._run_forever_setup()
    try:
        while True:
            self._run_once()
            if self._stopping:
                break
    finally:
        self._run_forever_cleanup()

非常直观！run_forever 方法就是一个无线的 While 循环。它调用 _run_once() 方法来处理所有事务（包括 Event 和 Task，这个后面会讲到），根据 _stopping 判断是否需要停止。

所以，可以说 Event loop 就是一个 while 循环，因为后面提到的所有任务都是在这个 while 循环里面发生的，所以更合适的说法应该叫调度器。

1.4 在 Event Loop 中注册（schedule）回调

使用 call_soon() 来注册 Callback 函数，Event loop 会尽快调用 Callback。

Schedule the callback to be called with args arguments at the next iteration of the event loop.

使用 call_later() 可以在指定延迟后调用 Callback。

Schedule callback to be called after the given delay number of seconds (can be either an int or a float).

import datetime
import asyncioloop = asyncio.get_event_loop()
start_time = datetime.datetime.now().timestamp()
def print_time():print("Elapsed time:", datetime.datetime.now().timestamp() - start_time)loop.call_soon(print_time)  # Schedule print_time to be called soon
loop.call_soon(print_time)  # Schedule print_time to be called soon again
loop.call_later(1, print_time)  # Schedule print_time to be called after 1 second
loop.run_until_complete(asyncio.sleep(1))  # Run the event loop for 1 second

输出：

Elapsed time: 0.0
Elapsed time: 0.0
Elapsed time: 1.0129709243774414

1.5 递归的艺术

在异步编程中，我们有时需要创建一个持续不断的周期性任务。由于 call_later 只能执行一次，开发者常常利用 Trampoline（蹦床） 技术：在函数执行结束前，将自己再次注册到 Event Loop 中。
这种模式比 while True 更优雅，因为它不会霸占 CPU，而是执行完一次后就“弹”回到 Loop 的调度队列里，静候下一次触发。

import datetime
import asyncio
loop = asyncio.get_event_loop()
def trampoline():print("Trampoline called at:", datetime.datetime.now())loop.call_later(1, trampoline)  # Schedule itself to be called again after 1 secondloop.call_soon(trampoline)  # Start the trampoline
loop.run_forever()  # Run the event loop indefinitely

输出

Trampoline called at after 0.0 seconds
Trampoline called at after 1.003988 seconds
Trampoline called at after 2.006745 seconds
Trampoline called at after 3.019901 seconds
...

1.6 致命阻塞

CPU 密集型任务是异步编程中非常严重的一个问题，这会使得整个事件循环卡死在某处，其他所有任务都会被停止。

import datetime
import asyncio
loop = asyncio.get_event_loop()
def trampoline():print("Trampoline called at:", datetime.datetime.now())loop.call_later(1, trampoline)  # Schedule itself to be called again after 1 seconddef hog_cpu():print("Hogging CPU at:", datetime.datetime.now())# Simulate a blocking operationend_time = datetime.datetime.now() + datetime.timedelta(seconds=2)while datetime.datetime.now() < end_time:passprint("Finished hogging CPU at:", datetime.datetime.now())loop.call_soon(trampoline) # Start the trampoline
loop.call_later(5, hog_cpu)  # Schedule another CPU hogging after 5 seconds
loop.run_forever()  # Run the event loop indefinitely

输出：

Trampoline called at: 2024-06-01 12:00:00.
Trampoline called at: 2024-06-01 12:00:01.
Trampoline called at: 2024-06-01 12:00:02.
Hogging CPU at: 2024-06-01 12:00:05
Finished hogging CPU at: 2024-06-01 12:00:07.
Trampoline called at: 2024-06-01 12:00:08.
Trampoline called at: 2024-06-01 12:00:09.
...

在这种情况下，当 hog_cpu 函数被调用时，它会阻塞 Event Loop 2 秒，导致 Trampoline 在此期间无法被调用。