1. TCP服务器实现方案
1.1 基础TCP服务器(阻塞模式)
import socket
def create_tcp_server(host='127.0.0.1', port=8888):
# 创建TCP套接字
server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
# 设置端口重用
server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
# 绑定地址
server_socket.bind((host, port))
# 开始监听
server_socket.listen(5)
return server_socket
1.2 非阻塞模式
# 设置非阻塞模式
server_socket.setblocking(False)
try:
client_socket, client_addr = server_socket.accept()
except BlockingIOError:
pass # 没有新连接时继续执行
1.3 Select模式
import select
# 创建监听列表
inputs = [server_socket]
outputs = []
while True:
readable, writable, exceptional = select.select(inputs, outputs, inputs)
for sock in readable:
if sock is server_socket:
client_socket, client_addr = sock.accept()
inputs.append(client_socket)
else:
try:
data = sock.recv(1024)
if data:
# 处理数据
pass
else:
inputs.remove(sock)
sock.close()
except Exception:
inputs.remove(sock)
sock.close()
1.4 Poll模式
import select
# 创建poll对象
poll_obj = select.poll()
# 注册服务器套接字
poll_obj.register(server_socket.fileno(), select.POLLIN)
while True:
events = poll_obj.poll()
for fd, event in events:
if fd == server_socket.fileno():
client_socket, client_addr = server_socket.accept()
poll_obj.register(client_socket.fileno(), select.POLLIN)
1.5 Epoll模式
import select
# 创建epoll对象
epoll = select.epoll()
# 注册服务器套接字
epoll.register(server_socket.fileno(), select.EPOLLIN)
while True:
events = epoll.poll()
for fd, event in events:
if fd == server_socket.fileno():
client_socket, client_addr = server_socket.accept()
epoll.register(client_socket.fileno(), select.EPOLLIN)
2. IO多路复用对比
特性 |
Select |
Poll |
Epoll |
连接数限制 |
1024 |
无限制 |
无限制 |
检测方式 |
轮询 |
轮询 |
事件通知 |
效率 |
低 |
中 |
高 |
系统支持 |
全平台 |
Linux |
Linux |
3. 文件描述符
3.1 基本概念
- 文件描述符是非负整数
- 是内核维护的文件记录表索引
- Linux/Unix特有概念
3.2 标准文件描述符
import sys
# 标准输入
stdin_fd = sys.stdin.fileno() # 0
# 标准输出
stdout_fd = sys.stdout.fileno() # 1
# 标准错误
stderr_fd = sys.stderr.fileno() # 2
4. 协程
4.1 基本概念
- 微线程/纤程
- 用户态的轻量级线程
- 程序员自主控制的并发单元
4.2 特点对比
特性 |
线程 |
协程 |
切换成本 |
高 |
低 |
调度方式 |
系统调度 |
用户调度 |
并发性能 |
中 |
高 |
编程复杂度 |
中 |
较高 |
4.3 适用场景
# IO密集型任务
async def io_task():
await asyncio.sleep(1)
# IO操作
# 计算密集型任务(不推荐)
def cpu_task():
for i in range(1000000):
# 计算操作
pass
4.4 最佳实践
- IO密集型任务优先使用协程
- 计算密集型任务使用多进程
- 避免协程中包含阻塞操作
- 合理设计协程调度策略
5. 性能优化建议
5.1 网络服务器
- 使用适合的IO多路复用机制
- 合理设置缓冲区大小
- 实现超时机制
- 做好错误处理
5.2 并发处理
- IO密集型:协程 > 多线程 > 多进程
- CPU密集型:多进程 > 多线程 > 协程
- 混合场景:考虑多进程+协程的方案