200字
ReActor模式原理探索
2026-06-22
2026-06-22
学习总结

ReActor原理介绍

Reactor是什么

一种处理高并发网络请求的设计模式，核心特征：

  • 同步非阻塞：IO 操作不挂起线程，没就绪立刻返回
  • 事件驱动：把"IO 处理"转化为"就绪事件处理"
  • 解耦：事件监听分发 与 事件处理 分离

其主要解决的问题：避免为每个连接开独立线程，减少资源消耗。

问题 原因
内存爆炸 线程栈默认 1MB，10k 连接 = 10GB 内存
上下文切换开销 线程阻塞/唤醒涉及内核态切换，大量连接时切换次数剧增
缓存失效 线程切换导致 CPU 缓存频繁失效
锁竞争 共享数据需要加锁，线程越多竞争越激烈

笔记

多线程适合 IO 任务，但密集网络请求反而是单线程事件驱动更高效。线程阻塞会占用内存和调度资源，从挂起到唤醒涉及多次内核态/用户态切换。而单线程配合 epoll 等系统调用，让内核统一监视所有连接的就绪状态，数据到了直接处理，没有线程上下文切换开销，也没有大量线程栈的内存占用。Redis 单线程抗高并发、Netty 的主从 Reactor 模型，都是基于此思路——用系统调用代替线程做"等待"，把线程资源留给真正的计算处理。

核心处理流程

注册事件 → 事件监听与分发 → 事件处理
   ↑________________________|

使用 IO 多路复用（select/poll/epoll）解决"什么时候 IO 就绪"的问题

以 Linux epoll API 为例：

epoll_ctl  →  epoll_wait  →  callback
(注册fd)      (阻塞等事件)     (处理IO: accept/read/write)

服务端网络 IO 流程（阻塞 vs 非阻塞）

步骤 操作
1 socket()
2 bind()
3 listen()
4 accept()
5 read/recv()
6 write/send()
7 循环 5,6 直到 read 返回 0（对端关闭）
8 close()

阻塞 IO 的问题：不知道数据什么时候来，只能阻塞线程等。高并发时线程数爆炸。
Reactor 的解决：一个线程监视所有 fd，谁就绪处理谁。

参考epoll可以实现一个非阻塞的IO处理流程
Linux IO模式及 select、poll、epoll详解(转载)

简易实现

以网络io为例逐步实现一个简易的Reactor示例

基础实现（阻塞）

import socket
import time

def main() -> None:
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind(("0.0.0.0", 9999))
    server.listen(128)
    print("[START] 阻塞服务器跑在 :9999 ...")

    while True:
        # 阻塞等待连接
        conn, addr = server.accept()
        print(f"[ACCEPT] 新连接: {addr}")

        while True:
            # 阻塞等待数据
            data = conn.recv(1024)
            if not data:
                print(f"[CLOSE] {addr}")
                conn.close()
                break
          
            print(f"[RECV]  {addr}: {data!r}")
            conn.send(b"echo: " + data)

if __name__ == "__main__":
    main()

调用Select基础实现（非阻塞）

import selectors
import socket
import time

sel = selectors.DefaultSelector()

def accept(server: socket.socket) -> None:
    conn, addr = server.accept()
    print(f"[ACCEPT] 新连接: {addr}")
    sel.register(conn, selectors.EVENT_READ, data=read)

def read(conn: socket.socket) -> None:
    data = conn.recv(1024)
    if data:
        print(f"[RECV]  {conn.getpeername()}: {data!r}")
        conn.send(b"echo: " + data) # 原样回显
    else:
        print(f"[CLOSE] {conn.getpeername()}")
        sel.unregister(conn)
        conn.close()

def main() -> None:
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind(("0.0.0.0", 9999))
    server.listen(128) # 监听
    server.setblocking(False) # 关键!非阻塞
    print("[START] Reactor 跑在 :9999 ...")

    sel.register(server, selectors.EVENT_READ, data=accept) # 注册 server socket 到 selector
  
    try:
        while True:
            events = sel.select(timeout=None) # 阻塞等待事件发生
            for key, mask in events:
                callback = key.data # 获取回调函数
                callback(key.fileobj) # 调用回调函数
    except KeyboardInterrupt:
        print("\n[STOP] 收到 Ctrl+C,退出")
    finally:
        sel.close()
        server.close()


if __name__ == "__main__":
    main()

用python的select库取代了默认的socket阻塞，默认阻塞假如有10个连接同时接入，需要依次阻塞等待激活顺次接收执行。使用select后由主线程统一阻塞等待，有事件则调用回调执行继续等待

聊天室案例（非阻塞）

import selectors
import socket
from collections import deque

sel = selectors.DefaultSelector()

# conn -> deque[bytes]  每个连接的发送队列
send_queues: dict[socket.socket, deque] = {}

def broadcast(msg: bytes, exclude=None):
    """O(n) 入队，不阻塞。真正的 send 等 selector 通知写就绪"""
    for conn in list(send_queues):
        if conn is exclude:
            continue
        send_queues[conn].append(msg)
        # 注册写事件（如果还没注册）
        sel.modify(conn, selectors.EVENT_READ | selectors.EVENT_WRITE, data=handle)

def handle(conn: socket.socket, mask: int=0):
    """统一回调，根据 mask 分流"""
    if mask & selectors.EVENT_READ:
        on_read(conn)
    if mask & selectors.EVENT_WRITE:
        on_write(conn)

def on_read(conn: socket.socket):
    try:
        data = conn.recv(1024)
    except OSError as e:
        print(f"[ERROR] {conn.getpeername()} 连接异常断开 {e}\n")
        broadcast(f"[SYS] {conn.getpeername()} 异常离开\n".encode())
        cleanup(conn)
        return
    if data:
        # 公屏广播，排除自己
        broadcast(f"[{conn.getpeername()}] {data.decode()}".encode(), exclude=conn)
    else:
        # 断开
        broadcast(f"[SYS] {conn.getpeername()} 离开\n".encode())
        cleanup(conn)

def on_write(conn: socket.socket):
    q = send_queues[conn]
    while q:
        msg = q.popleft()
        try:
            sent = conn.send(msg)  # 非阻塞，可能只发一部分
            if sent < len(msg):
                q.appendleft(msg[sent:])  # 没发完的塞回去
                break  # 等下次写就绪
        except BlockingIOError:
            q.appendleft(msg)  # 塞回去，等下次
            break
        except OSError as e:
            print(f"[ERROR] {conn.getpeername()} 连接在发送数据时异常断开 {e}\n")
            broadcast(f"[SYS] {conn.getpeername()} 异常离开\n".encode())
            cleanup(conn)
            return
    else:
        # 队列空了，取消写事件，只留读
        sel.modify(conn, selectors.EVENT_READ, data=handle)

def accept(server: socket.socket, mask: int=0):
    conn, addr = server.accept()
    conn.setblocking(False)
    send_queues[conn] = deque()
    sel.register(conn, selectors.EVENT_READ, data=handle)
    broadcast(f"[SYS] {addr} 加入\n".encode())

def cleanup(conn: socket.socket):
    send_queues.pop(conn, None)
    sel.unregister(conn)
    conn.close()

def main():
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind(("0.0.0.0", 9999))
    server.listen(128)
    server.setblocking(False)
    sel.register(server, selectors.EVENT_READ, data=accept)
  
    print("[START] Reactor 跑在 :9999 ...")
  
    try:
        while True:
            events = sel.select(timeout=None)
            for key, mask in events:
                callback = key.data # 获取回调函数
                callback(key.fileobj, mask) # 调用回调函数
    except KeyboardInterrupt:
        print("\n[STOP]")
    finally:
        for conn in list(send_queues):
            conn.close()
        sel.close()
        server.close()

if __name__ == "__main__":
    main()

案例改造（用asyncio的默认事件驱动实现替换）

import selectors
import socket
from collections import deque
import asyncio

# 所有活跃连接
clients: set[asyncio.StreamWriter] = set()

async def broadcast(msg: bytes, exclude: asyncio.StreamWriter = None):
    for client in clients:
        if client is not exclude:
            try:
                client.write(msg)
                await asyncio.wait_for(client.drain(), timeout=0.5)  # 等待发送完成，避免过快发送导致连接断开
            except (ConnectionResetError, BrokenPipeError, asyncio.TimeoutError):
                # 慢客户端/已断开的,踢掉
                clients.discard(client)
                print(f"[ERROR] {client.get_extra_info('peername')} 连接异常接收超时，现已触发断开")
                try:
                    client.close()
                except Exception:
                    print(f"[ERROR] 关闭连接时发生异常")

async def handle_client(reader: asyncio.StreamReader, writer: asyncio.StreamWriter):
    """每个客户端连接对应一个协程，负责处理这个连接的读写"""
    addr = writer.get_extra_info('peername')
    print(f"[ACCEPT] 新连接: {addr}")
    clients.add(writer)
    await broadcast(f"[SYS] {addr} 加入了聊天室".encode(), exclude=writer)
    try:
        while True:
            data = await reader.read(1024)
            if not data:
                print(f"[CLOSE] {addr}")
                break
            print(f"[RECV] {addr}: {data!r}")
            # 广播给其他客户端的逻辑需要自己实现，这里先简单回显
            await broadcast(f"{addr}: ".encode() + data, exclude=writer)
    except ConnectionResetError:
        print(f"[ERROR] {addr} 连接异常断开")
    finally:
        clients.discard(writer)
        await broadcast(f"[SYS] {addr} 离开了聊天室\n".encode(), exclude=writer)
        writer.close()

async def main():
    server = await asyncio.start_server(handle_client, "0.0.0.0", 9999)
    print("[START] Chatroom Server (asyncio) 跑在 :9999 ...")
    async with server:
        await server.serve_forever()

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        print("\n[STOP]")

附件

client客户端实现：

import socket
import threading

def receve_thread(sock: socket.socket):
    while True:
        data = sock.recv(1024)
        if not data:
            print("[DISCONNECTED] 服务器已断开连接")
            sock.close()
            break
        print(f"{data.decode()}")

def main() -> None:
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
        sock.connect(("localhost", 9999))
        print("[CONNECTED] 已连接到服务器")

        threading.Thread(target=receve_thread, args=(sock,), daemon=True).start()
        try:
            while True:
                msg = input()
                if msg.lower() == "exit":
                    print("[EXIT] 退出客户端")
                    break
                sock.send(msg.encode())
        except (KeyboardInterrupt, BrokenPipeError):
            print("\n[EXIT] 退出客户端")

    

if __name__ == "__main__":
    main()
ReActor模式原理探索
Author
Administrator
Published at
2026-06-22
License
CC BY-NC-SA 4.0
Linux IO模式及 selec...
Transformer位置编码(正...

Comment

© Andy'Home. All Rights Reserved. / RSS / Sitemap
Powered by Halo & Theme-Fuwari