Python的事件循环库

YaoQi

发布于 2026-07-10 19:05:11

什么是事件？

比如定时器到期;文件、网络套接字等I/O可读，可写，进程信号（如SIGINT）到达等。

什么是事件循环？

你可以把事件循环（Event Loop） 想象成一个无限循环的“调度中心”或“指挥官”。它的核心任务非常简单，以I/O操作为例：

不断地检查两个队列：一个是要执行的任务队列，另一个是刚刚完成的I/O操作结果队列。
协调调度：从任务队列中取出一个任务来执行。
- 如果这个任务需要执行一个I/O操作（比如从网络下载数据、读写文件、访问数据库），事件循环不会傻傻地等待它完成。它会立刻将这个I/O操作丢给操9作系统去处理，同时把这个任务挂起，然后立刻去执行队列里的下一个任务。
- 当操作系统完成那个I/O操作后，会把结果放回一个队列。事件循环在下一轮循环中会检查到这个结果，然后唤醒刚才被挂起的那个任务，并把结果交给它，让它继续执行。

它的最大优势：用单线程实现高并发。

在一个传统的多线程程序中，每个连接都需要一个线程，线程的创建、切换和销毁开销很大。而事件循环在单个线程内通过“任务切换”的方式处理海量的I/O操作。在I/O等待期间，CPU不会空闲，而是去执行其他可以执行的任务，从而极大地提高了程序的效率和吞吐量，特别适合处理网络请求、文件操作等I/O密集型场景。

Python 有哪些事件循环库？

基于 asyncio（主流首选）

asyncio（标准库）
- Python 官方事件循环与协程框架，生态最大（aiohttp、uvicorn、httpx、FastAPI 等都以此为基础）。
uvloop
- 高性能的 asyncio 事件循环实现，作为插入式替换使用，网络服务场景常见提速。
- 用法：

import asyncio, uvloop
asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
asyncio.run(main())

Tornado（IOLoop）

5.0 起默认基于 asyncio 运行，保留自身 IOLoop 接口（Jupyter 体系大量使用）。

AnyIO
- 异步抽象层，统一 asyncio/Trio 接口；库作者常用（Starlette/FastAPI/HTTPX 使用它）。

非 asyncio（自带或不同模型的事件循环/并发框架）

Trio
- 结构化并发（nursery、友好的取消/超时语义），自带事件循环；可用 trio-asyncio 与 asyncio 互操作。
Curio
- 也是结构化并发思路的实验性框架，学习价值大，生态较小。
Twisted（Reactor）
- 老牌回调/Deferred 模型，协议栈丰富；可装 asyncio reactor 与现代生态互通。
gevent（libev + greenlet）
- 绿色线程与猴子补丁的同步风格并发，历史项目众多；与 asyncio 不直接兼容。
eventlet
- 类 gevent 的绿色线程库（OpenStack 早期大量使用）。
pyuv
- libuv 的 Python 绑定，偏底层，适合需要直接操控 libuv 的场景或做自定义高性能循环。

GUI 专用/与 asyncio 的桥接

Qt（PyQt/PySide）：QEventLoop；用 qasync 将 Qt 事件循环与 asyncio 打通。
GLib/GTK（PyGObject）：GLib MainLoop；可用 gbulb 等桥接到 asyncio。
Tkinter/wxPython/Kivy：各自有 main loop；也有社区桥接方案，但不如 Qt/GLib 成熟。

事件循环库示例

让事件循环管理任务，首先要设计任务，任务是可以执行完就退出，或者循环不退出的。

然后事件循环库添加任务，最后执行.run() 函数启动事件循环，线程就在事件循环库的调度下运行了，主程序就阻塞在那了, 等待任务执行完。

asyncio的简单示例：

import asyncio

##  单个任务
async def hello_sleep_world():
    print('Hello ...')
    await asyncio.sleep(1)
    print('... World!')

asyncio.run(hello_sleep_world())
#
# Hello ...
# ... World!

## 多个任务
async def main():
    return await asyncio.gather(
        hello_sleep_world(),
        hello_sleep_world(),
        hello_sleep_world()
    )

asyncio.run(main())
# Hello ...
# Hello ...
# Hello ...
# ... World!
# ... World!
# ... World!

GUI库Thinker示例：

import tkinter as tk
from tkinter import ttk


class ProgressBarApp:
    def __init__(self, root):
        self.root = root
        self.root.title("进度条示例")
        self.is_running = False

        # 进度条
        self.progress = ttk.Progressbar(root, length=300, mode='determinate')
        self.progress.pack(pady=20)

        # 百分比标签
        self.percent_label = tk.Label(root, text="0%")
        self.percent_label.pack()

        # 按钮框架
        self.btn_frame = tk.Frame(root)
        self.btn_frame.pack(pady=10)

        # 开始按钮
        self.start_btn = tk.Button(self.btn_frame, text="开始", command=self.start_progress)
        self.start_btn.pack(side=tk.LEFT, padx=5)

        # 暂停按钮
        self.pause_btn = tk.Button(self.btn_frame, text="暂停", command=self.pause_progress, state=tk.DISABLED)
        self.pause_btn.pack(side=tk.LEFT, padx=5)

        # 复位按钮
        self.reset_btn = tk.Button(self.btn_frame, text="复位", command=self.reset_progress)
        self.reset_btn.pack(side=tk.LEFT, padx=5)

    def start_progress(self):
        self.is_running = True
        self.start_btn.config(state=tk.DISABLED)
        self.pause_btn.config(state=tk.NORMAL)
        self.update_progress()

    def pause_progress(self):
        self.is_running = False
        self.start_btn.config(state=tk.NORMAL)
        self.pause_btn.config(state=tk.DISABLED)

    def update_progress(self):
        if self.is_running and self.progress['value'] < 100:
            self.progress['value'] += 5
            self.percent_label.config(text=f"{self.progress['value']}%")
            self.root.after(500, self.update_progress)
        elif self.progress['value'] >= 100:
            self.is_running = False
            self.start_btn.config(state=tk.DISABLED)
            self.pause_btn.config(state=tk.DISABLED)

    def reset_progress(self):
        self.is_running = False
        self.progress['value'] = 0
        self.percent_label.config(text="0%")
        self.start_btn.config(state=tk.NORMAL)
        self.pause_btn.config(state=tk.DISABLED)


root = tk.Tk()
app = ProgressBarApp(root)
root.mainloop()

本文参与腾讯云自媒体同步曝光计划，分享自微信公众号。

原始发表：2025-09-11，如有侵权请联系 cloudcommunity@tencent.com 删除

python