Python 并发编程 多路复用IO模型详解

一、什么是多路复用IO模型

在传统的 I/O 模型中，当一个线程或者进程要进行 I/O 操作的时候，会阻塞当前的任务，等待 I/O 完成后才能继续执行后续的任务。这种模式既浪费时间，也浪费资源，无法高效地利用 CPU。

多路复用 IO 模型是一种更加高效的 I/O 处理模型，在这种模式下，可以实现多个 I/O 任务的并发处理，避免了每个 I/O 任务的等待，提高了 CPU 的利用效率。常见的多路复用 IO 模型有 select、poll、epoll。

二、select 模块

1. 简介

select 是一种比较古老的多路复用 IO 模型，在 Python 中，它被封装在 select 模块下。select 的主要作用是监听文件描述符，当有一个文件描述符准备就绪以后，就会返回该文件描述符，从而实现多个 I/O 任务的并发处理。

2. select.select()

select.select() 方法的基本用法如下：

import select
import socket

server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server_socket.bind(("127.0.0.1", 8888))
server_socket.listen()

inputs = [server_socket]

while True:
    readable, _, _ = select.select(inputs, [], [])
    for sock in readable:
        if sock == server_socket:
            client_socket, address = server_socket.accept()
            inputs.append(client_socket)
        else:
            data = sock.recv(1024)
            if not data:
                inputs.remove(sock)
                sock.close()
                continue
            msg = f"received message: {data.decode()}"
            sock.sendall(msg.encode())

在这个示例中，我们首先初始化一个 server_socket，将其加入 inputs 列表中。然后创建一个死循环，调用 select.select(inputs, [], []) 监听 inputs 中的文件描述符，如果有文件描述符准备就绪，就进入 for 循环进行处理。

如果准备就绪的文件描述符是 server_socket，那么我们就执行 server_socket.accept() 方法，创建一个新的套接字 client_socket，并将其加入 inputs 列表中。如果准备就绪的文件描述符是 client_socket，那么我们就执行 sock.recv(1024) 方法，接收客户端发送的数据，并返回一条响应消息。

3. select.poll()

select.poll() 方法的用法和 select.select() 类似，不同的是，select.poll() 更加高效，可以在象征性的文件描述符数量下应用于数以百万计的节点。

import select
import socket

server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server_socket.bind(("127.0.0.1", 8888))
server_socket.listen()

poll = select.poll()
poll.register(server_socket, select.POLLIN)

fd_to_socket = {server_socket.fileno(): server_socket}

while True:
    events = poll.poll(10000)
    for fd, event in events:
        if fd == server_socket.fileno():
            client_socket, address = server_socket.accept()
            poll.register(client_socket, select.POLLIN)
            fd_to_socket[client_socket.fileno()] = client_socket
        elif event & select.POLLIN:
            data = fd_to_socket[fd].recv(1024)
            if not data:
                poll.unregister(fd)
                fd_to_socket[fd].close()
                del fd_to_socket[fd]
                continue
            msg = f"received message: {data.decode()}"
            fd_to_socket[fd].sendall(msg.encode())

在这个示例中，我们首先初始化一个 poll 对象，将 server_socket 注册到 poll 中。然后创建一个字典 fd_to_socket，将 server_socket 的文件描述符和对象存入该字典中。

接下来，我们进入死循环，调用 poll.poll(10000) 监听 poll 中的文件描述符，如果有文件描述符准备就绪，就进入 for 循环进行处理。

如果准备就绪的文件描述符是 server_socket，那么我们就执行 server_socket.accept() 方法，创建一个新的套接字 client_socket，并将其加入 poll 中，同时在 fd_to_socket 中创建该套接字的键值对。如果准备就绪的文件描述符是 client_socket，那么我们就执行 fd_to_socket[fd].recv(1024) 方法，接收客户端发送的数据，并返回一条响应消息。

三、epoll 模块

epoll 是一个比较新的多路复用 IO 模型，在 Python 中，它被封装在 epoll 模块下。epoll 是 Linux 2.6 内核提供的新的 IO 多路复用接口，旨在替代旧的 select/poll，相比旧的方式，epoll 能够有效地处理大量并发链接，性能更加优异。

import select
import socket

server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
server_socket.bind(("127.0.0.1", 8888))
server_socket.listen()

epoll = select.epoll()
epoll.register(server_socket.fileno(), select.EPOLLIN)

fd_to_socket = {server_socket.fileno(): server_socket}

while True:
    events = epoll.poll(1)
    for fd, event in events:
        if fd == server_socket.fileno():
            client_socket, address = server_socket.accept()
            epoll.register(client_socket.fileno(), select.EPOLLIN)
            fd_to_socket[client_socket.fileno()] = client_socket
        elif event & select.EPOLLIN:
            data = fd_to_socket[fd].recv(1024)
            if not data:
                epoll.unregister(fd)
                fd_to_socket[fd].close()
                del fd_to_socket[fd]
                continue
            msg = f"received message: {data.decode()}"
            fd_to_socket[fd].sendall(msg.encode())

在这个示例中，我们首先初始化一个 epoll 对象，将 server_socket 注册到 epoll 中。然后创建一个字典 fd_to_socket，将 server_socket 的文件描述符和对象存入该字典中。

接下来，我们进入死循环，调用 epoll.poll(1) 监听 epoll 中的文件描述符，如果有文件描述符准备就绪，就进入 for 循环进行处理。

如果准备就绪的文件描述符是 server_socket，那么我们就执行 server_socket.accept() 方法，创建一个新的套接字 client_socket，并将其加入 epoll 中，同时在 fd_to_socket 中创建该套接字的键值对。如果准备就绪的文件描述符是 client_socket，那么我们就执行 fd_to_socket[fd].recv(1024) 方法，接收客户端发送的数据，并返回一条响应消息。

四、总结

多路复用 IO 模型是实现高效多任务的关键之一，Python 提供了 select 和 epoll 模块，可以帮助我们实现非阻塞 IO 的应用程序。我们需要根据具体的应用场景和需求，选取合适的 I/O 模型。