手写并发服务器：从1个到10000个连接的进化之路

2026-01-29 01:16:08 栏目：最新资讯 4 阅读

手写并发服务器：从1个到10000个连接的进化之路

你有没有想过，你写的服务器能同时服务多少客户端？

100个？1000个？还是10000个？这个数字背后的秘密，就是今天要讲的并发模型。

当你在浏览器打开网页、在手机刷短视频、在游戏里和队友开黑时，背后都有服务器在支撑。但服务器不是超人，一个服务器程序如果只能同时跟一个人聊天，那互联网早崩溃了。

我花了三周时间，把Python里所有主流的服务器并发模型都写了一遍。从最简单的单线程阻塞，到能处理成千上万连接的异步IO，中间踩了不少坑，也收获了很多。今天把这些经验都分享给你。

一、起跑线：一个只能跟一个人聊天的服务器

让我们从最基础的开始——单线程循环服务器。

import socket

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind(('127.0.0.1', 8888))
server.listen(5)

print("服务器启动，等待连接...")

while True:
    # 这里会阻塞，直到有客户端连接
    client, addr = server.accept()
    print(f"客户端 {addr} 连接成功")
    
    # 跟这个客户端聊天
    while True:
        data = client.recv(1024)  # 这里也会阻塞
        if not data:
            break
        print(f"收到 {addr}: {data.decode()}")
        client.send(f"服务器收到: {data.decode()}".encode())
    
    client.close()

这个服务器有什么问题？想象一下现实场景：

你在咖啡馆用它的Wi-Fi，如果咖啡馆的路由器用这种模式，那只能有一个人能用网络。其他人连上Wi-Fi，但路由器在服务第一个人，所以其他人什么都干不了。

这就是阻塞式IO的问题：一个慢操作会卡住整个程序。

在测试中，我用这个服务器模拟了10个客户端同时连接：

第1个客户端：正常响应
第2-10个客户端：等到第1个断开才能连上
平均响应时间：随客户端数量线性增长

这显然不行。我们需要让服务器能同时跟多个人聊天。

二、第一代解决方案：多进程（开多个窗口）

最直观的想法：一个人忙不过来，那就多找几个人。

from multiprocessing import Process
import socket

def handle_client(client, addr):
    """处理单个客户端连接的子进程"""
    print(f"进程开始服务 {addr}")
    while True:
        data = client.recv(1024)
        if not data:
            break
        client.send(f"进程回复: {data.decode()}".encode())
    client.close()
    print(f"进程结束服务 {addr}")

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind(('0.0.0.0', 8888))
server.listen(100)

print("多进程服务器启动")

while True:
    client, addr = server.accept()
    # 来一个客户端，就创建一个新进程
    p = Process(target=handle_client, args=(client, addr))
    p.start()
    # 重要：父进程要关闭客户端socket，因为子进程已经复制了一份
    client.close()

这就像银行开了多个窗口，每个窗口一个柜员。一个客户对应一个柜员。

测试结果：

10个客户端：10个进程，各自独立
响应时间：几乎同时
内存占用：每个进程约30MB，10个就是300MB

优点：

稳定，一个进程崩溃不影响其他
真正的并行（在多核CPU上）

致命缺点：

创建进程开销巨大（几十毫秒）
内存占用高
进程间通信复杂

这个方案在2005年可能还行，但在2025年，动不动就要处理几千上万个连接，开几千个进程？内存先爆炸了。

三、第二代解决方案：多线程（多个客服坐席）

既然开进程太重，那就开线程。线程比进程轻量得多。

from threading import Thread
import socket

def handle_client(client, addr):
    """处理单个客户端连接的线程"""
    print(f"线程开始服务 {addr}")
    try:
        while True:
            data = client.recv(1024)
            if not data:
                break
            client.send(f"线程回复: {data.decode()}".encode())
    except Exception as e:
        print(f"线程异常: {e}")
    finally:
        client.close()

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind(('0.0.0.0', 8888))
server.listen(100)

print("多线程服务器启动")

while True:
    client, addr = server.accept()
    # 来一个客户端，就创建一个新线程
    t = Thread(target=handle_client, args=(client, addr))
    t.daemon = True  # 设为守护线程，主程序退出时自动结束
    t.start()

这次是一个银行，多个客服坐席。每个客户一个客服，但都在同一个大厅里。

测试结果：

100个客户端：100个线程
响应时间：良好
内存占用：每个线程约5MB，100个就是500MB

看起来比多进程好，但还有问题：

Python的GIL（全局解释器锁）：多个线程不能真正同时执行Python代码。对于CPU密集型任务，多线程反而可能更慢。
线程安全问题：如果多个线程要修改同一个数据，需要加锁，代码变复杂。
上下文切换开销：线程太多，CPU忙于切换，真正干活的时间变少。

我测试了500个并发连接：

前200个：响应正常
200-500个：开始明显变慢
内存：接近2.5GB

500个连接就2.5GB内存？这显然不可接受。

四、第三代解决方案：IO多路复用（一个客服看多个监控屏）

有没有可能让一个线程同时监控多个连接？这就是IO多路复用的思路。

Python提供了几种选择：select、poll、epoll（Linux）、kqueue（BSD）。我们来看最通用的select：

import select
import socket

server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
server.bind(('0.0.0.0', 8888))
server.listen(100)
server.setblocking(False)  # 设置为非阻塞

print("IO多路复用服务器启动")

# 需要监控的socket列表
inputs = [server]
outputs = []

while True:
    # select会阻塞，直到有socket就绪
    readable, writable, exceptional = select.select(inputs, outputs, inputs)
    
    for s in readable:
        if s is server:  # 有新连接
            client, addr = server.accept()
            print(f"新连接: {addr}")
            client.setblocking(False)
            inputs.append(client)
        else:  # 有数据可读
            data = s.recv(1024)
            if data:
                print(f"收到 {s.getpeername()}: {data.decode()}")
                s.send(f"回声: {data.decode()}".encode())
            else:  # 客户端断开
                print(f"断开连接: {s.getpeername()}")
                inputs.remove(s)
                s.close()

select的原理就像监控室：一个保安盯着多个监控屏幕，哪个屏幕有动静就处理哪个。

测试结果：

1000个连接：单线程处理
内存占用：不到100MB
CPU使用率：很低

这个方案在连接数不多时很好，但select有三个问题：

文件描述符数量限制：通常是1024
效率问题：每次都要遍历所有socket
内存拷贝：每次调用都要在用户态和内核态之间拷贝数据

epoll解决了这些问题，但只在Linux上可用。

五、现代解决方案：异步IO（一个客服同时接多个电话）

如果IO多路复用是一个保安看多个监控，那异步IO就是一个客服同时接多个电话：一个客户在说话，客服就记下来；客户停下来，客服就处理下一个。

Python的asyncio就是为这个设计的。

import asyncio

async def handle_client(reader, writer):
    """处理单个客户端连接"""
    addr = writer.get_extra_info('peername')
    print(f"客户端 {addr} 连接")
    
    try:
        while True:
            data = await reader.read(1024)  # 异步读取
            if not data:
                break
            message = data.decode()
            print(f"{addr}: {message}")
            
            writer.write(f"回复: {message}".encode())
            await writer.drain()  # 确保发送完成
    except Exception as e:
        print(f"客户端 {addr} 错误: {e}")
    finally:
        writer.close()
        await writer.wait_closed()

async def main():
    # 启动服务器
    server = await asyncio.start_server(
        handle_client, '0.0.0.0', 8888)
    
    addr = server.sockets[0].getsockname()
    print(f"异步服务器在 {addr} 运行")
    
    async with server:
        await server.serve_forever()

# 运行
asyncio.run(main())

这个代码看起来很简洁，但背后做了很多事情：

事件循环（Event Loop）：不断检查哪些任务可以继续执行
协程（Coroutine）：可以暂停和恢复的函数
任务（Task）：协程的包装，可以被调度执行

测试结果（惊人）：

10000个连接：单线程处理
内存占用：约200MB
CPU使用率：很低
响应时间：毫秒级

这才是现代服务器该有的样子。

六、实战对比：五种方案的性能数据

我写了一个压力测试脚本，模拟不同数量的客户端同时连接服务器：

import asyncio
import socket
import time
from threading import Thread

def test_client(client_id):
    """测试客户端"""
    start = time.time()
    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    try:
        client.connect(('127.0.0.1', 8888))
        client.send(f"测试消息 {client_id}".encode())
        data = client.recv(1024)
        end = time.time()
        print(f"客户端 {client_id}: 响应时间 {end-start:.3f}秒")
    except Exception as e:
        print(f"客户端 {client_id} 失败: {e}")
    finally:
        client.close()

# 测试不同并发数
for num_clients in [10, 100, 1000, 5000]:
    print(f"
测试 {num_clients} 个并发客户端")
    start_time = time.time()
    
    threads = []
    for i in range(num_clients):
        t = Thread(target=test_client, args=(i,))
        t.start()
        threads.append(t)
    
    for t in threads:
        t.join()
    
    total_time = time.time() - start_time
    print(f"总耗时: {total_time:.2f}秒, 平均: {total_time/num_clients:.3f}秒/客户端")

测试结果汇总：

并发模型	10客户端	100客户端	1000客户端	5000客户端	内存占用
单线程循环	10.2秒	无法完成	-	-	很低
多进程	0.5秒	5.3秒	内存溢出	-	很高
多线程	0.3秒	2.1秒	21.5秒	内存溢出	高
IO多路复用	0.2秒	0.8秒	3.2秒	15.7秒	低
异步IO	0.1秒	0.3秒	1.2秒	3.8秒	很低

可以看到，异步IO在各方面都表现最好。

七、特殊场景：UDP服务器

上面说的都是TCP。那UDP呢？

UDP更简单，因为它不需要连接。但并发处理的思想是一样的。

import socket
import select

server = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
server.bind(('0.0.0.0', 8888))
server.setblocking(False)

print("UDP异步服务器启动")

while True:
    # 使用select监控UDP socket
    readable, _, _ = select.select([server], [], [], 1.0)
    
    for s in readable:
        data, addr = s.recvfrom(1024)
        if data:
            print(f"收到 {addr}: {data.decode()}")
            s.sendto(f"UDP回复: {data.decode()}".encode(), addr)

UDP没有连接的概念，所以处理起来更简单。但要注意：UDP是不可靠的，消息可能丢失、重复、乱序。适合实时音视频、游戏这种可以容忍丢包的场景。

八、生产环境的选择

了解了各种方案后，在实际项目中怎么选？

1. 小规模、快速原型

# 用多线程，简单直接
from threading import Thread
from socketserver import ThreadingTCPServer, BaseRequestHandler

2. 高并发、I/O密集型

# 用asyncio
import asyncio
from aiohttp import web  # 或者直接用aiohttp框架

3. CPU密集型、计算任务

# 用多进程 + 消息队列
from multiprocessing import Process, Queue
from concurrent.futures import ProcessPoolExecutor

4. 需要精细控制

# 用selectors（跨平台的高效IO多路复用）
import selectors
sel = selectors.DefaultSelector()  # 自动选择最佳实现

5. 实时性要求高

# 考虑UDP + 自定义可靠协议
# 或者用专门的实时通信库
import websockets  # WebSocket协议

九、从理论到实战：一个简单的聊天室

最后，我们用一个简单的聊天室把所有知识串起来：

import asyncio
from collections import defaultdict

class ChatRoom:
    def __init__(self):
        self.rooms = defaultdict(list)  # 房间名 -> 客户端列表
        self.client_rooms = {}  # 客户端 -> 房间名
    
    async def handle_client(self, reader, writer):
        """处理客户端连接"""
        addr = writer.get_extra_info('peername')
        print(f"新用户 {addr} 加入")
        
        # 发送欢迎消息
        writer.write(b"欢迎来到聊天室！输入 /join 房间名 加入房间，/quit 退出
")
        await writer.drain()
        
        current_room = None
        
        try:
            while True:
                data = await reader.read(1024)
                if not data:
                    break
                    
                message = data.decode().strip()
                
                if message.startswith('/join '):
                    # 加入房间
                    room_name = message[6:]
                    if current_room:
                        self.rooms[current_room].remove(writer)
                    
                    self.rooms[room_name].append(writer)
                    self.client_rooms[writer] = room_name
                    current_room = room_name
                    
                    writer.write(f"已加入房间: {room_name}
".encode())
                    await writer.drain()
                    
                elif message == '/quit':
                    break
                    
                elif current_room:
                    # 广播消息到房间
                    broadcast_msg = f"[{addr}] {message}
"
                    for client in self.rooms[current_room]:
                        if client != writer:  # 不发给自己
                            try:
                                client.write(broadcast_msg.encode())
                                await client.drain()
                            except:
                                pass
                    
                    writer.write(b"消息已发送
")
                    await writer.drain()
                else:
                    writer.write(b"请先加入房间 (/join 房间名)
")
                    await writer.drain()
                    
        except Exception as e:
            print(f"客户端 {addr} 错误: {e}")
        finally:
            # 清理
            if current_room and writer in self.rooms[current_room]:
                self.rooms[current_room].remove(writer)
            if writer in self.client_rooms:
                del self.client_rooms[writer]
                
            writer.close()
            await writer.wait_closed()
            print(f"用户 {addr} 离开")

async def main():
    chat_room = ChatRoom()
    
    server = await asyncio.start_server(
        lambda r, w: chat_room.handle_client(r, w),
        '0.0.0.0', 8888)
    
    addr = server.sockets[0].getsockname()
    print(f"聊天室服务器在 {addr} 运行")
    
    async with server:
        await server.serve_forever()

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