如何在Python中实现Proactor模式？能否实现真正的Proactor风格I/O处理？

一、Windows平台：基于IOCP的原生Proactor实现 #

Windows的I/O完成端口（IOCP）是原生的Proactor风格异步I/O机制，Python的asyncio模块已直接支持对应的ProactorEventLoop——内核后台处理I/O操作，完成后触发回调，无需用户空间主动轮询就绪状态，完全符合你所说的「真正的」Proactor处理逻辑。

实现步骤 #

1. 替换默认事件循环为ProactorEventLoop
2. 编写标准asyncio异步代码，底层自动复用IOCP能力

代码示例（异步Echo服务器） #

import asyncio

async def handle_echo(reader, writer):
    data = await reader.read(100)
    message = data.decode()
    addr = writer.get_extra_info('peername')
    print(f"Received {message!r} from {addr!r}")

    print(f"Sending {message!r}")
    writer.write(data)
    await writer.drain()

    print("Close the connection")
    writer.close()
    await writer.wait_closed()

async def main():
    # 强制启用ProactorEventLoop
    loop = asyncio.ProactorEventLoop()
    asyncio.set_event_loop(loop)
    
    server = await asyncio.start_server(
        handle_echo, '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())

二、Linux平台：基于io_uring的Proactor风格实现 #

Linux没有原生IOCP，但Linux 5.1+引入的io_uring是一款接近Proactor模式的异步I/O接口：内核负责发起和处理I/O操作，完成后主动通知用户空间，符合「真正的」Proactor特性。Python生态有两种主流实现方式：

方案1：Python 3.12+标准库内置IO_UringEventLoop #

Python 3.12开始在asyncio中内置了基于io_uring的事件循环，直接支持Proactor风格处理：

import asyncio

async def main():
    # 切换到IO_UringEventLoop
    loop = asyncio.IO_UringEventLoop()
    asyncio.set_event_loop(loop)
    
    # 后续编写标准asyncio异步代码即可，底层自动复用io_uring
    print("Running with io_uring Proactor loop")

if __name__ == '__main__':
    asyncio.run(main())

方案2：第三方库pyuring（兼容低版本Python） #

如果使用Python 3.12以下版本，可以通过pyuring直接调用io_uring接口，手动实现Proactor逻辑：

import pyuring
import socket

def on_accept(fd, args):
    client_fd, addr = socket.fromfd(fd, socket.AF_INET, socket.SOCK_STREAM)
    print(f"Accepted connection from {addr}")
    # 注册读事件
    ring.submit_read(client_fd, bytearray(1024), user_data=(client_fd, 'read'))

def on_read(fd, data, args):
    client_fd, _ = args
    if data:
        print(f"Received: {data.decode().strip()}")
        # 回写数据
        ring.submit_write(client_fd, data, user_data=(client_fd, 'write'))
    else:
        client_fd.close()

ring = pyuring.IOUring()
server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server_sock.bind(('127.0.0.1', 8888))
server_sock.listen(5)
server_sock.setblocking(False)

# 注册accept事件
ring.submit_accept(server_sock.fileno(), user_data=(server_sock.fileno(), 'accept'))

while True:
    ring.submit()
    # 等待I/O完成并处理回调
    for completion in ring.wait():
        if completion.user_data[1] == 'accept':
            on_accept(completion.res, completion.user_data)
        elif completion.user_data[1] == 'read':
            on_read(completion.user_data[0], completion.data, completion.user_data)
        elif completion.user_data[1] == 'write':
            # 写完后继续注册读事件
            ring.submit_read(completion.user_data[0], bytearray(1024), user_data=completion.user_data)

三、核心结论 #

• 「真正的」Proactor模式核心是内核主动处理I/O操作并通知完成，而非用户空间轮询I/O就绪状态（Reactor模式）。Windows的IOCP和Linux的io_uring都满足这一核心特性，对应的Python实现属于真正的Proactor风格处理。
• 跨平台场景可通过判断操作系统动态选择事件循环：Windows用ProactorEventLoop，Linux（5.1+）用IO_UringEventLoop或pyuring，其他平台 fallback 到Reactor模式的SelectorEventLoop。

内容的提问来源于stack exchange，提问作者luminousmen