聊一聊 Python 的 socket，以及 select、poll、epoll 又是怎么一回事？

import socket
# socket.socket() 会返回一个「主动套接字」
server = socket.socket(
    # 表示使用 IPv4，如果是 socket.AF_INET6
    # 则表示使用 IPv6
    socket.AF_INET,
    # 表示建立 TCP 连接，如果是 socket.SOCK_DGRAM
    # 则表示建立 UDP 连接
    socket.SOCK_STREAM
)
# 当然这两个参数也可以不传，因为默认就是它
# 设置套接字属性，这里让端口释放后立刻就能再次使用
server.setsockopt(socket.SOL_SOCKET,
                  socket.SO_REUSEADDR, True)
# 将「主动套接字」绑定在某个 IP 和端口上
server.bind(("localhost", 12345))
# 监听，此时「主动套接字」会变成「监听套接字」
# 里面的参数表示 backlog，代表的含义后面说
server.listen(5)
# 调用 accept，等待客户端连接，此时会阻塞在这里
# 如果客户端连接到来，那么会返回「已连接套接字」，也就是这里的 conn
# 至于 addr 则是一个元组，保存了客户端连接的信息（IP 和端口）
conn, addr = server.accept()
# 下面我们通过「已连接套接字」conn 和客户端进行消息的收发
# 收消息使用 recv、发消息使用 send，和 read、write 本质是一样的
while True:
    msg = conn.recv(1024)
    # 当客户端断开连接时，msg 会收到一个空字节串
    if not msg:
        print("客户端已经断开连接")
        conn.close()
        break
    print("客户端发来消息:", msg.decode("utf-8"))
    # 然后我们加点内容之后，再给客户端发过去
    conn.send("服务端收到, 你发的消息是: ".encode("utf-8") + msg)

import socket
# 返回主动套接字
client = socket.socket(socket.AF_INET,
                       socket.SOCK_STREAM)
# 连接服务端
client.connect(("localhost", 12345))
while True:
    # 发送消息
    data = input("请输入内容: ")
    if data.strip().lower() in ("q", "quit", "exit"):
        client.close()
        print("Bye~~~")
        break
    client.send(data.encode("utf-8"))
    print(client.recv(1024).decode("utf-8"))

import socket
import threading
server = socket.socket()
server.setsockopt(socket.SOL_SOCKET,
                  socket.SO_REUSEADDR, True)
server.bind(("localhost", 12345))
server.listen(5)
def handle_message(conn, addr):
    while True:
        msg = conn.recv(1024)
        if not msg:
            print(f"客户端(ip: {addr[0]}, port: {addr[1]}) 已经断开连接")
            conn.close()
            break
        print(f"客户端(ip: {addr[0]}, port: {addr[1]}) 发来消息:",
              msg.decode("utf-8"))
        conn.send("服务端收到, 你发的消息是: ".encode("utf-8") + msg)
while True:
    conn, addr = server.accept()
    threading.Thread(
        target=handle_message,
        args=(conn, addr)
    ).start()

import socketserver
# 自定义一个类，必须继承 BaseRequestHandler
class ServiceHandler(socketserver.BaseRequestHandler):
    """
    内部提供了三个重要属性
        self.request: 已连接套接字 conn
        self.client_address: 客户端信息 addr
        self.server: 服务端实例（一会我们会创建它）
    然后我们必须要实现 handle 方法，处理客户端连接时会自动调用
    此外还有两个方法，分别是 setup 和 finish，实不实现均可
    """
    def setup(self) -> None:
        """在执行 handle 之前调用，用于提前做一些连接相关的设置"""
    def finish(self) -> None:
        """在执行 handle 之后调用，用于资源释放等等"""
        self.request.close()
    def handle(self) -> None:
        """
        处理客户端连接
        这里的 self.request 就相当于之前的 conn
        """
        client_ip, client_port = self.client_address
        while True:
            msg = self.request.recv(1024)
            if not msg:
                print(f"客户端(ip: {client_ip}, port: {client_port}) 已经断开连接")
                self.request.close()
                break
            print(f"客户端(ip: {client_ip}, port: {client_port}) 发来消息:",
                  msg.decode("utf-8"))
            self.request.send("服务端收到, 你发的消息是: ".encode("utf-8") + msg)
# 绑定 IP 和端口，以及用于处理的 Handler
# 这里的 ThreadingTCPServer 实例就是 ServiceHandler 里面的 self.server
server = socketserver.ThreadingTCPServer(
    ("localhost", 12346),
    ServiceHandler
)
# 开启无限循环，监听连接
server.serve_forever()
# 如果关闭监听，那么调用 server.shutdown()

server = socket.socket()
...
server.listen(5)

import socket
server = socket.socket()
server.bind(("localhost", 12345))
# 调用 setblocking 方法，传入 False
# 表示将监听套接字和已连接套接字的类型设置为非阻塞
server.setblocking(False)
server.listen(5)
while True:
    try:
        # 非阻塞的监听套接字调用 accept() 时
        # 如果发现没有客户端连接，则会立刻抛出 BlockingIOError
        # 因此这里写了个死循环
        conn, addr = server.accept()
    except BlockingIOError:
        pass
    else:
        break
while True:
    try:
        # 同理，非阻塞的已连接套接字在调用 recv() 时
        # 如果发现客户端没有发数据，那么同样会报错
        msg = conn.recv(1024)
    except BlockingIOError:
        pass
    else:
        print(msg.decode("utf-8"))
        conn.send(b"data from server")

import select
"""
select 函数接收四个参数
    rlist：一个列表，监听那些可读的 socket
    wlist：一个列表，监听那些可写的 socket
    xlist：一个列表，监听那些出错的 socket
    timeout：超时时间
"""
select.select()

import socket
import select
from queue import Queue
from typing import Dict
server = socket.socket()
server.bind(("localhost", 12345))
server.setsockopt(socket.SOL_SOCKET,
                  socket.SO_REUSEADDR, True)
# 必须设置为非阻塞，IO 多路复用要搭配非阻塞套接字
server.setblocking(False)
server.listen(5)
# 以上我们就创建了监听套接字，它负责监听是否有客户端连接
# 所以当事件发生时，属于可读事件，代表客户端连接过来了
# 所以 server 应该放在 rlist 里面
rlist = [server]
wlist = []
xlist = rlist
# 因为可以监听多个套接字，所以 rlist、wlist、xlist 都是列表
# 但在初始状态下，select 只需要监听一个套接字（server）即可
message_queues: Dict[socket.socket, Queue] = {}
while True:
    # 开启 select 监听，返回三个元素
    # readable：   rlist 中发生可读事件的 socket
    # writeable：  wlist 中发生可写事件的 socket
    # exceptional：xlist 中发生异常的 socket
    readable, writeable, exceptional = select.select(rlist, wlist, xlist)
    # 遍历 readable
    for r in readable:
        # 如果 r is server，则代表监听套接字有事件发生
        # 显然是客户端连接来了
        if r is server:
            # 监听套接字是非阻塞的，那么已连接套接字默认也是非阻塞的
            # 当然你也可以调用 conn.setblocking(False) 显式地设置一下
            conn, addr = server.accept()
            print(f"和客户端 {addr[0]}:{addr[1]} 建立连接")
            # 将已连接套接字添加到 rlist 中，让它也被 select 监听
            # 当客户端发消息时，它会进入活跃状态，有事件发生
            # 然后被 select 监测到，放到 readable 中，这样遍历的时候就可以处理它了
            rlist.append(conn)
            # 由于客户端连接之后要发消息，那么我们是不是要将消息保存起来呢？
            message_queues[conn] = Queue()  # 负责保存后续客户端发的消息
        else:
            # 如果 r is not server，则代表是已连接套接字有事件发生
            # 说明是某个客户端发送消息了，我们要处理它
            data = r.recv(1024)
            if data:
                # 这里的 r 就是活跃的已连接套接字，调用它的 getpeername 方法
                # 也可以获取到客户端连接的 ip 和 端口
                addr = r.getpeername()
                print(f"收到客户端 {addr[0]}:{addr[1]} 发来的消息:",
                      f"`{data.decode('utf-8')}`")
                # message_queues 保存了所有的已连接套接字
                # 每一个套接字都对应一个队列，找到该活跃套接字对应的队列
                message_queues[r].put(data)  # 将消息放进去
                # 消息放进去了，服务端是不是也要回复呢？显然这属于可写事件
                # 我们还要将 r 放到 wlist 中，这样 select 就会监测到
                # 然后将它放到 writeable 中，我们遍历的时候就可以处理它了
                if r not in wlist:
                    wlist.append(r)
            else:
                # 走到这里说明 data 为假，说明客户端断开连接了，发来一个 b''
                addr = r.getpeername()
                print(f"客户端 {addr[0]}:{addr[1]} 已断开连接")
                # 我们要将套接字从 rlist、wlist 当中移除
                # 客户端都断开连接了，那么 select 也就不需要再监听了
                rlist.remove(r)
                if r in wlist:
                    wlist.remove(r)
                # 对了，还要将它从 message_queues 里面移除
                message_queues.pop(r)
                r.close()  # 关闭套接字连接
    # 以上是遍历可读事件，可读事件可以发生在监听套接字上面（和客户端建立连接）
    # 也可以发生在已连接套接字上面（客户端发信息了）
    # 如果没有事件发生或者处理完毕，那么接下来就要遍历可写事件
    # 而可写事件一定发生在已连接套接字上面（要回消息给客户端）
    for w in writeable:
        message_queue = message_queues[w]
        # 一开始队列里面肯定是有消息的，因为我们手动往里面放了一条
        # 但如果队列为空，说明服务端已经回复过了，那么要将该套接字从 wlist 里面移除
        # 让它变为非活跃状态，不再满足可写
        # 等到下一次客户端发消息时，再将它添加到 wlist 中，让它变得可写
        if message_queue.empty():
            wlist.remove(w)
            continue
        # 获取消息
        data = message_queue.get()
        # 发送给客户端
        w.send("服务端收到，你发的消息是: ".encode("utf-8") + data)
    # 然后遍历 xlist，如果在跟某个客户端通信的过程中，出现了错误
    # 那么将套接字从 rlist、wlist、xlist、message_queue 当中都删除
    # 然后再关闭套接字连接
    for x in exceptional:
        addr = x.getpeername()
        print(f"和客户端 {addr[0]}:{addr[1]} 通信出现错误")
        rlist.remove(x)
        if x in wlist:
            wlist.remove(x)
        message_queues.pop(x)
        x.close()

import select
from select import POLLIN, POLLOUT, POLLERR, POLLHUP
# select.poll 是一个类
# 我们实例化一个 poll 对象
poll = select.poll()
# 给指定的套接字绑定事件
# 第一个参数可以是描述符，也可以是套接字
# 第二个参数是要绑定的事件
# POLLIN：可读事件（rlist）
# POLLOUT：可写事件（wlist）
# POLLERR：出现异常（xlist）
# POLLHUP：连接中断
poll.register(..., POLLIN | POLLERR)
# 取消某个套接字的事件绑定
poll.unregister(...)

import socket
import select
from select import POLLIN, POLLOUT, POLLERR, POLLHUP
from typing import Dict
from queue import Queue
server = socket.socket()
server.bind(("localhost", 12345))
server.setsockopt(socket.SOL_SOCKET,
                  socket.SO_REUSEADDR, True)
server.setblocking(False)
server.listen(5)
# 通过文件描述符找到套接字
fd2sk = {server.fileno(): server}
# 保存套接字接收到的客户端发来的消息
message_queues: Dict[int, Queue] = {}
# 实例化一个 poll 对象
poll = select.poll()
# 首先要对 server 进行注册，正如我们使用 select.select 时
# 要先将 server 放在 rlist 里面，然后事件为可读
# 不过由于可能发生错误，因此事件类型为 POLLIN | POLLERR
poll.register(server, POLLIN | POLLERR)
# 开启无限循环
while True:
    # poll 方法接收一个 timeout，不传则表示不设置超时
    # 它和 select.select 的第四个参数的含义相同
    # 当有事件发生时，会返回相应的文件描述符和事件
    events = poll.poll() # 正式开启监听
    # 我们进行遍历
    for fd, event in events:
        # 说明是监听套接字活跃了
        if fd == server.fileno():
            conn, addr = fd2sk[fd].accept()
            print(f"和客户端 {addr[0]}:{addr[1]} 建立连接")
            # 对 conn 进行注册，下一轮循环的时候也会对它进行监听
            # 这里可以传文件描述符、也可以传套接字
            # 如果传套接字，那么会自动调用 fileno 获取描述符
            poll.register(conn, POLLIN | POLLHUP | POLLERR)
            # 维护文件描述符到套接字的映射
            fd2sk[conn.fileno()] = conn
            # 为每个文件描述符维护一个队列，用于保存客户端发来的消息
            message_queues[conn.fileno()] = Queue()
        # 否则说明是已连接套接字有事件发生
        # 那么事件是可读还是可写呢？显然要通过 event 判断
        elif event & POLLIN:  # 可读
            data = fd2sk[fd].recv(1024)
            if data:  # 有数据
                addr = fd2sk[fd].getpeername()
                print(f"收到客户端 {addr[0]}:{addr[1]} 发来的消息:",
                      f"`{data.decode('utf-8')}`")
                # 客户端发消息，服务端也要回消息
                # 因此要给它注册一个可写事件
                poll.register(fd, POLLOUT | POLLHUP | POLLERR)
                # 然后将消息保存起来
                message_queues[fd].put(data)
            else:  # 客户端断开连接
                addr = fd2sk[fd].getpeername()
                print(f"客户端 {addr[0]}:{addr[1]} 已断开连接")
                # 取消监听，会将所有事件全部取消
                poll.unregister(fd)
                # 关闭连接
                fd2sk[fd].close()
                # 从字典中移除
                fd2sk.pop(fd)
        elif event & POLLOUT:  # 已连接套接字可写
            message_queue = message_queues[fd]
            if message_queue.empty():
                # 队列为空，说明消息已经发出去了
                # 那么套接字就不再可写了，因此要取消监听
                # 等到下一次客户端发消息时，再变得可写
                poll.unregister(fd)
                # 但 unregister 会取消所有事件的监听
                # 因此还要重新注册可读事件
                # 不然后续客户端发消息时，就无法处理了
                poll.register(fd, POLLIN | POLLHUP | POLLERR)
            else:
                data = message_queue.get()
                fd2sk[fd].send(
                    "服务端收到，你发的消息是: ".encode("utf-8") + data
                )
        elif event & POLLERR:  # 发生错误
            addr = fd2sk[fd].getpeername()
            print(f"和客户端 {addr[0]}:{addr[1]} 通信出现错误")
            poll.unregister(fd)
            fd2sk[fd].close()
            message_queues.pop(fd)

import select
import socket
from queue import Queue
from select import (
    # epoll 和 poll 的用法相似
    # 把事件换成 epoll 的事件即可
    EPOLLIN,
    EPOLLOUT,
    EPOLLERR,
    EPOLLHUP
)
from typing import Dict
server = socket.socket()
server.bind(("localhost", 12345))
server.setsockopt(socket.SOL_SOCKET,
                  socket.SO_REUSEADDR, True)
server.setblocking(False)
server.listen(5)
fd2sk = {server.fileno(): server}
message_queues: Dict[int, Queue] = {}
# 实例化一个 epoll 对象
epoll = select.epoll()
# 给 server 注册读事件
epoll.register(server, EPOLLIN | EPOLLERR)
while True:
    # 仍然是调用 poll 方法，开始轮询
    events = epoll.poll()
    for fd, event in events:
        if fd == server.fileno():
            conn, addr = fd2sk[fd].accept()
            print(f"和客户端 {addr[0]}:{addr[1]} 建立连接")
            # 给已连接套接字注册读事件
            # 第一个参数可以传文件描述符、也可以传套接字
            epoll.register(conn, EPOLLIN | EPOLLHUP | EPOLLERR)
            fd2sk[conn.fileno()] = conn
            message_queues[conn.fileno()] = Queue()
        elif event & EPOLLIN:  # 可读
            data = fd2sk[fd].recv(1024)
            if data:  # 有数据
                addr = fd2sk[fd].getpeername()
                print(f"收到客户端 {addr[0]}:{addr[1]} 发来的消息:",
                      f"`{data.decode('utf-8')}`")
                # 客户端发消息了，那么套接字要回复消息，因此满足可写
                # 但是和 poll 不同，epoll 只能给一个套接字注册一次
                # 而之前已经注册过一次了（已连接套接字创建时，注册了读事件）
                # 因此再注册就会报错，因为不能连续注册
                # 所以我们需要将 fd 上的事件取消掉，然后重新注册
                epoll.unregister(fd)
                # 重新注册，此时要同时注册读事件和写事件
                epoll.register(fd, EPOLLIN | EPOLLOUT | EPOLLHUP | EPOLLERR)
                message_queues[fd].put(data)
            else:
                # 客户端断开连接了
                addr = fd2sk[fd].getpeername()
                print(f"客户端 {addr[0]}:{addr[1]} 已断开连接")
                epoll.unregister(fd)
                fd2sk[fd].close()
                fd2sk.pop(fd)
                
        elif event & EPOLLOUT:  # 可写
            message_queue = message_queues[fd]
            if message_queue.empty():
                # 队列是空的，说明消息已经发走了，那么应该取消写事件
                # 做法也很简单：先将事件全部取消，然后重新注册读事件
                # 但也可以通过 modify 方法，直接修改事件类型
                epoll.modify(fd, EPOLLIN | EPOLLHUP | EPOLLERR)
            else:
                data = message_queue.get()
                fd2sk[fd].send(
                    "服务端收到，你发的消息是: ".encode("utf-8") + data
                )
        elif event & EPOLLERR:
            addr = fd2sk[fd].getpeername()
            print(f"和客户端 {addr[0]}:{addr[1]} 通信出现错误")
            epoll.unregister(fd)
            fd2sk[fd].close()
            message_queues.pop(fd)

int epoll_create(int size);

int epoll_ctl(int epfd, int op, int fd, 
              struct epoll_event *event);

int epoll_wait(int epfd, struct epoll_event *events, 
               int maxevents, int timeout);

import socket
from queue import Queue
# selectors 里面提供了多种"多路复用器"
# 除了 select、poll、epoll 之外
# 还有 kqueue，这个是针对 BSD 平台的
try:
    from selectors import (
        SelectSelector,
        PollSelector,
        EpollSelector,
        KqueueSelector
    )
except ImportError:
    pass
# 由于种类比较多，所以提供了 DefaultSelector
# 会根据当前的系统种类，自动选择一个合适的多路复用器
from selectors import (
    DefaultSelector,
    EVENT_READ,  # 读事件
    EVENT_WRITE,  # 写事件
)
server = socket.socket()
server.bind(("localhost", 12345))
server.setsockopt(socket.SOL_SOCKET,
                  socket.SO_REUSEADDR, True)
server.setblocking(False)
server.listen(5)
message_queues = {}
# 实例化一个多路复用器
sel = DefaultSelector()
def accept(server: socket.socket):
    """和客户端建立连接"""
    conn, addr = server.accept()
    print(f"和客户端 {addr[0]}:{addr[1]} 建立连接")
    # 一旦建立连接，那么就要接收客户端消息
    # 所以我们要绑定事件，register 方法接收三个参数
    # 参数一：传一个套接字即可
    # 参数二：事件类型，这里是读事件
    # 参数三：事件发生时，执行的回调函数
    sel.register(conn, EVENT_READ, recv)
    # 表示当 conn 可读时，就去执行 recv 函数
    message_queues[conn] = Queue()
def recv(conn: socket.socket):
    """接收客户端数据"""
    data = conn.recv(1024)
    addr = conn.getpeername()
    if data:  # 有数据
        print(f"收到客户端 {addr[0]}:{addr[1]} 发来的消息:",
              f"`{data.decode('utf-8')}`")
        # 收到数据了，那么要给客户端回复，所以要绑定可写事件
        # 让事件可写，当事件发生时，执行 send 函数
        sel.modify(conn, EVENT_READ | EVENT_WRITE, send)
        message_queues[conn].put(data)
    else:
        print(f"客户端 {addr[0]}:{addr[1]} 已断开连接")
        conn.close()
        # 取消监听
        sel.unregister(conn)
        message_queues.pop(conn)
def send(conn: socket.socket):
    """给客户端发送数据"""
    message_queue = message_queues[conn]
    if message_queue.empty():
        # 队列为空说明已经发送过了，将事件改成可读
        # 继续监听客户端发来的消息
        sel.modify(conn, EVENT_READ, recv)
    else:
        data = message_queue.get()
        conn.send(
            "服务端收到，你发的消息是: ".encode("utf-8") + data
        )
# 给监听套接字注册可读事件
# 当有客户端连接，去执行 accept 函数
sel.register(server, EVENT_READ, accept)
# 在 accept 函数里面创建已连接套接字 conn
# 然后给 conn 绑定可读事件，当客户端发消息时，去执行 recv 函数
# 在 recv 函数里面给套接字绑定可写事件，然后去执行 send 函数
while True:
    # 内部会根据操作系统，选择一个合适的多路复用
    events = sel.select()
    # key 是一个 namedtuple
    # 内部有如下字段：'fileobj', 'fd', 'events', 'data'
    # key.fd 就是套接字的文件描述符
    # key.fileobj 则是套接字本身
    # key.data 是给套接字绑定的回调
    # key.event 则是事件
    for key, mask in events:
        # 事件发生时，获取回调，然后调用
        # 回调显然就是这里的 accept、recv、send
        callback = key.data 
        callback(key.fileobj)