Python Asyncio：IO多路复用与事件循环的调度策略
大家好！今天我们来深入探讨Python asyncio框架中的核心概念：IO多路复用以及事件循环中不同调度策略的对比，特别是Epoll/Kqueue和Proactor模式。理解这些概念对于编写高性能、可扩展的异步应用程序至关重要。
1. 异步编程与IO密集型任务
在传统的同步编程模型中，程序执行流程是线性的，每个操作必须等待前一个操作完成后才能开始。对于IO密集型任务，例如网络请求、文件读写等，程序的大部分时间都花费在等待IO操作完成上，导致CPU空闲，资源利用率低下。
异步编程则允许程序在等待IO操作完成的同时执行其他任务。当IO操作完成后，程序会收到通知，然后继续处理IO结果。这种方式可以显著提高CPU利用率，提升程序的并发性能。
2. IO多路复用：解决并发IO的关键
IO多路复用是一种允许单个线程同时监听多个文件描述符（例如socket）的技术。当其中任何一个文件描述符准备好进行IO操作时（例如可读、可写），内核会通知应用程序。应用程序可以根据通知，选择性地处理已经准备好的文件描述符，而无需阻塞等待。
常见的IO多路复用技术包括：
-
Select: 最早的IO多路复用技术，由POSIX标准定义。它通过轮询文件描述符集合来检测IO事件。Select的缺点是每次调用都需要将整个文件描述符集合从用户空间复制到内核空间，并且支持的文件描述符数量有限（通常是1024）。
-
Poll: 类似于Select，但解决了Select的文件描述符数量限制问题。Poll使用一个
pollfd结构体数组来表示需要监听的文件描述符，并且支持更大的文件描述符数量。然而，Poll仍然需要轮询所有文件描述符来检测IO事件。 -
Epoll: Linux特有的IO多路复用技术，提供了更高的效率和可扩展性。Epoll使用红黑树来存储需要监听的文件描述符，并且只在文件描述符状态发生变化时才通知应用程序。这避免了不必要的轮询，提高了性能。Epoll支持两种工作模式：LT (Level Triggered) 和 ET (Edge Triggered)。
-
Kqueue: FreeBSD、macOS等系统提供的IO多路复用技术，类似于Epoll。Kqueue也使用事件队列来管理需要监听的文件描述符，并且提供了丰富的事件类型和过滤选项。
3. 事件循环（Event Loop）：异步编程的核心
事件循环是异步编程的核心组成部分。它负责监听IO事件，并根据事件类型调度相应的回调函数。事件循环通常包含以下几个步骤：
-
选择（Select/Poll/Epoll/Kqueue）: 监听所有注册的文件描述符，等待IO事件发生。
-
获取事件: 从内核获取已发生的IO事件。
-
调度回调函数: 根据事件类型，执行相应的回调函数。这些回调函数通常是协程（coroutines）。
-
循环: 返回第一步，继续监听IO事件。
4. Asyncio的事件循环与IO多路复用
Asyncio的事件循环底层使用了操作系统提供的IO多路复用机制。在Linux系统中，Asyncio通常使用Epoll；在FreeBSD/macOS系统中，Asyncio通常使用Kqueue；在Windows系统中，Asyncio使用IOCP (I/O Completion Ports)。
Asyncio提供了一系列API来注册文件描述符和回调函数，例如：
-
loop.add_reader(fd, callback, *args): 注册一个文件描述符，当可读时执行回调函数。 -
loop.add_writer(fd, callback, *args): 注册一个文件描述符，当可写时执行回调函数。 -
loop.create_task(coroutine): 创建一个任务，将其添加到事件循环中。
5. Proactor模式 vs Reactor模式
理解Asyncio的调度策略，需要区分两种常见的异步编程模式：Reactor模式和Proactor模式。
| Feature | Reactor模式 | Proactor模式 |
|---|---|---|
| IO操作发起者 | 应用程序 | 操作系统 |
| 数据读写 | 应用程序负责 | 操作系统负责 |
| 回调函数执行时机 | 数据准备好后 | 数据读写完成后 |
| 优点 | 简单易懂，易于实现 | 性能更高，避免了数据在用户空间和内核空间之间的复制 |
| 缺点 | 需要应用程序处理IO操作，可能阻塞事件循环 | 实现复杂，需要操作系统提供更强的支持 |
-
Reactor模式： 应用程序注册感兴趣的事件（例如可读、可写），当事件发生时，事件循环通知应用程序，应用程序负责执行实际的IO操作（例如读取数据、写入数据）。Epoll和Kqueue通常与Reactor模式一起使用。
-
Proactor模式： 应用程序注册IO操作的完成事件，操作系统负责执行实际的IO操作，并在操作完成后通知应用程序。应用程序只需要处理IO操作的结果。Windows的IOCP是Proactor模式的典型实现。
5.1 Reactor模式的代码示例 (Epoll/Kqueue)
下面的代码展示了如何使用Asyncio和Reactor模式实现一个简单的TCP服务器：
import asyncio
async def handle_client(reader, writer):
addr = writer.get_extra_info('peername')
print(f"Accepted connection from {addr}")
while True:
try:
data = await reader.read(1024) # 等待客户端发送数据
if not data:
break
message = data.decode()
print(f"Received {message!r} from {addr}")
response = f"Server received: {message}".encode()
writer.write(response) # 发送响应
await writer.drain() # 等待数据发送完成
except ConnectionResetError:
print(f"Client {addr} disconnected abruptly.")
break
except Exception as e:
print(f"Error handling client {addr}: {e}")
break
print(f"Closing connection from {addr}")
writer.close()
await writer.wait_closed()
async def main():
server = await asyncio.start_server(
handle_client, '127.0.0.1', 8888
)
addr = server.sockets[0].getsockname()
print(f'Serving on {addr}')
async with server:
await server.serve_forever()
if __name__ == "__main__":
asyncio.run(main())在这个例子中，handle_client协程负责处理客户端连接。reader.read(1024)会挂起协程，直到有数据可读。事件循环会监听socket的可读事件，当有数据到达时，唤醒协程，读取数据，并发送响应。 这里利用了asyncio 默认的基于selector的事件循环，底层依赖的就是epoll或者kqueue。
5.2 模拟Proactor模式的代码示例
由于Python asyncio的标准库并没有提供直接的Proactor模式的实现（除了在Windows上使用IOCP），我们可以通过loop.run_in_executor来模拟Proactor模式。loop.run_in_executor允许我们将IO操作提交给一个线程池执行，从而避免阻塞事件循环。
import asyncio
import concurrent.futures
import time
async def read_file(loop, file_path):
def blocking_read(file_path):
print(f"Starting blocking read of {file_path} in thread...")
time.sleep(2) # Simulate a long-running IO operation
with open(file_path, 'r') as f:
content = f.read()
print(f"Finished blocking read of {file_path} in thread.")
return content
print(f"Submitting read of {file_path} to executor...")
content = await loop.run_in_executor(None, blocking_read, file_path) # None uses default ThreadPoolExecutor
print(f"File {file_path} content: {content[:50]}...") # Print first 50 characters
return content
async def main():
loop = asyncio.get_running_loop()
file_paths = ['file1.txt', 'file2.txt']
# Create dummy files
for file_path in file_paths:
with open(file_path, 'w') as f:
f.write(f"This is the content of {file_path}. It is a long string to simulate a substantial file.")
tasks = [read_file(loop, file_path) for file_path in file_paths]
results = await asyncio.gather(*tasks)
print("All file reads completed.")
for i, result in enumerate(results):
print(f"Result from {file_paths[i]}: {result[:20]}...") # Print first 20 chars
if __name__ == "__main__":
asyncio.run(main())在这个例子中，read_file协程使用loop.run_in_executor将文件读取操作提交给线程池。事件循环会继续执行其他任务，而不会被阻塞。当线程池中的线程完成文件读取操作后，事件循环会收到通知，并唤醒read_file协程，处理读取结果。 虽然使用了线程池，但其思想是模拟Proactor模式中由操作系统内核来执行IO操作，完成后通知应用。
6. Epoll的LT模式与ET模式
Epoll提供了两种工作模式：LT (Level Triggered) 和 ET (Edge Triggered)。
-
LT (Level Triggered): 只要文件描述符处于就绪状态（例如可读、可写），Epoll就会一直通知应用程序。这意味着，如果应用程序没有一次性读取完所有数据，下次调用Epoll时仍然会收到通知。
-
ET (Edge Triggered): 只有当文件描述符的状态发生变化时，Epoll才会通知应用程序。这意味着，如果应用程序没有一次性读取完所有数据，下次调用Epoll时可能不会收到通知，导致数据丢失。
| Feature | LT (Level Triggered) | ET (Edge Triggered) |
|---|---|---|
| 触发条件 | 文件描述符处于就绪状态 | 文件描述符状态发生变化 |
| 通知次数 | 只要就绪，就会一直通知 | 只有状态变化时才通知 |
| 编程复杂度 | 较低，易于使用 | 较高，需要更谨慎的处理 |
| 性能 | 较低，可能产生不必要的通知 | 较高，减少了系统调用 |
| 适用场景 | 简单的IO操作，对性能要求不高 | 高并发、高性能的IO密集型应用 |
在使用ET模式时，需要确保应用程序一次性读取完所有数据，或者使用非阻塞IO，并循环读取直到EAGAIN错误。
6.1 ET模式的代码示例
import asyncio
import socket
async def handle_connection(loop, sock):
addr = sock.getpeername()
print(f"Accepted connection from {addr}")
sock.setblocking(False) # 设置为非阻塞模式
while True:
try:
data = sock.recv(1024)
if not data:
break
message = data.decode()
print(f"Received {message!r} from {addr}")
response = f"Server received: {message}".encode()
sock.sendall(response) # 使用sendall确保所有数据都发送
except BlockingIOError:
# 没有数据可读，稍后重试
await asyncio.sleep(0.01) # Avoid busy-waiting
except ConnectionResetError:
print(f"Client {addr} disconnected abruptly.")
break
except Exception as e:
print(f"Error handling client {addr}: {e}")
break
print(f"Closing connection from {addr}")
sock.close()
async def main():
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
sock.bind(('127.0.0.1', 8888))
sock.listen(5)
sock.setblocking(False) # 设置为非阻塞模式
loop = asyncio.get_running_loop()
print('Serving on 127.0.0.1:8888')
while True:
try:
client_sock, addr = await loop.sock_accept(sock) # Use sock_accept to integrate with the event loop
loop.create_task(handle_connection(loop, client_sock))
except OSError as e:
if e.errno == 11: #errno.EAGAIN
await asyncio.sleep(0.01)
else:
print(f"Error accepting connection: {e}")
break
if __name__ == "__main__":
asyncio.run(main())在这个例子中，我们将socket设置为非阻塞模式，并在handle_connection协程中使用sock.recv读取数据。如果sock.recv返回BlockingIOError错误，表示没有数据可读，我们需要稍后重试。这确保了我们不会阻塞事件循环，并且能够正确处理ET模式下的IO事件。 这里需要额外注意的是，loop.sock_accept也使用了异步方式来接受连接，避免了阻塞。
7. Asyncio的Future和Task
在Asyncio中，Future表示一个异步操作的最终结果。Task是Future的一个子类，表示一个正在执行的协程。
-
Future可以被设置为已完成或未完成状态。当Future被设置为已完成状态时，可以获取其结果。 -
Task可以被取消。当Task被取消时，其对应的协程会收到CancelledError异常。
Asyncio提供了一系列API来操作Future和Task，例如：
-
asyncio.Future(): 创建一个Future对象。 -
asyncio.create_task(coroutine): 创建一个Task对象，并将其添加到事件循环中。 -
task.cancel(): 取消一个Task对象。 -
await future: 等待Future对象完成，并获取其结果。
8. 总结
理解IO多路复用（Epoll/Kqueue等）和事件循环的调度策略是编写高效Asyncio程序的关键。Reactor模式和Proactor模式代表了两种不同的异步编程思想，Asyncio主要基于Reactor模式实现，但可以通过loop.run_in_executor模拟Proactor模式。Epoll的LT模式和ET模式则提供了不同的IO事件通知机制，需要根据实际应用场景选择合适的模式。
希望今天的讲座能够帮助大家更好地理解Python Asyncio的底层原理，并能够编写出更加高性能、可扩展的异步应用程序。
更多IT精英技术系列讲座，到智猿学院