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

手写并发服务器：从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

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

我测试了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有三个问题：

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())

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

测试结果（惊人）：

• 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"\\n测试 {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}秒/客户端")

测试结果汇总：

可以看到，异步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 退出\\n")
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}\\n".encode())
await writer.drain()

elif message == '/quit':
break

elif current_room:
# 广播消息到房间
broadcast_msg = f"[{addr}] {message}\\n"
for client in self.rooms[current_room]:
if client != writer: # 不发给自己
try:
client.write(broadcast_msg.encode())
await client.drain()
except:
pass

writer.write(b"消息已发送\\n")
await writer.drain()
else:
writer.write(b"请先加入房间 (/join 房间名)\\n")
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())

这个聊天室用了异步IO，可以轻松支持上千人在线。你可以用telnet或写个简单的客户端来测试：

telnet 127.0.0.1 8888

总结

从单线程到异步IO，服务器的并发能力提升了几千倍。但选择哪种方案，还是要看具体需求：

• 学习、原型开发：从多线程开始，简单易懂
• 生产环境、高并发：用asyncio，性能最好
• 特殊需求：根据场景选择合适的技术栈