如果你已经走完了Python基础语法、函数、面向对象这些阶段开始看“网络编程”这一块那这套“Socket套接字 TCP开发 多进程”的组合拳基本是绕不开的。我自己带过不少新人发现大家最容易卡住的不是某一个API不会用而是这三样东西被拆成三章在学学了Socket不知道它和TCP什么关系学了TCP不知道多进程为什么配进来结果真到写一个能扛住并发请求的服务端时脑子里还是一团浆糊。这篇文章就按我实际开发中习惯的路径从Socket的原理讲到TCP服务端怎么写再一步步把单进程改成多进程每一段都有能跑的代码也把我在生产环境踩过的坑一并写出来。无论你是准备做后端接口、实时消息推送、物联网网关还是单纯想把Python网络编程这块地基打牢这篇文章都值得你按顺序看一遍。1. 为什么先搞懂Socket再谈高并发1.1 Socket套接字到底封装了什么很多初学者会把Socket、TCP、HTTP混在一起说面试时也经常被问“Socket和TCP是什么关系”。我先用一个生活里打电话的例子把这件事讲清楚。你打电话时需要先拨号对方接听然后你俩通话最后挂断。这个过程中“电话机”本身就是Socket而“电信网络里的通话协议规则”就是TCP。Python里的socket模块其实就是给你发了一台可以写程序的“电话机”底层那些网卡驱动、协议栈、路由转发它全都替你挡掉了。你只需要按照“拨号—接通—收发数据—挂断”的顺序去调用API就能完成一次网络通信。具体到代码层面客户端要做三件事创建套接字、发起连接、收发数据。服务端则是创建套接字、绑定端口、监听、接受连接、收发数据。就这么几个函数掌握之后你就能手写一个HTTP服务器。1.2 一次完整TCP请求在Python里长什么样拿一个最简单的HTTP请求来说当你在浏览器里访问一个网站时背后经历的是DNS解析拿到IP、TCP三次握手建立连接、发送HTTP请求、接收响应、四次挥手断开连接。这个过程在Python里用Socket模拟出来就是下面这段代码的核心逻辑。import socket # 创建TCP套接字AF_INET表示IPv4SOCK_STREAM表示TCP client socket.socket(socket.AF_INET, socket.SOCK_STREAM) # 发起TCP连接其实就是完成三次握手 client.connect((127.0.0.1, 8080)) # 发送数据sendall会等数据都发完才返回 client.sendall(bGET / HTTP/1.1\r\nHost: localhost\r\n\r\n) # 接收响应 response client.recv(4096) print(response.decode()) client.close()你看到没有核心就是socket()、connect()、sendall()、recv()、close()这几个方法。如果你手动跑过这段代码你会发现自己已经完成了一次完整的TCP通信。这也是我为什么建议进阶阶段先别急着上框架底层Socket一定要亲手写一遍否则后面学asyncio、Twisted、Tornado时你根本不知道它们替你省了多少事。2. 开发前的环境与调试准备2.1 环境选择与前置知识开发环境这块Python 3.8就可以我自己的项目长期在3.10和3.11上跑socket、multiprocessing这两个模块都是标准库不需要额外安装第三方包。至于编辑器用你顺手的就行VS Code、PyCharm、Vim都可以关键是要能直接跑Python脚本。有件事我提一句这篇文章里的代码都在Linux/macOS下验证过Windows下大部分代码也能跑但多进程的fork行为在Windows上不一样Windows默认使用spawn方式创建子进程所以如果你在Windows上想跑多进程TCP服务端最好把入口代码放到if __name__ __main__:里这个习惯是硬性的能避开很多诡异问题。2.2 三个高效的调试手段写网络程序最怕什么不是Bug是你看不到数据到底是怎么流动的。我推荐三个调试手段按效率排序第一telnet。这是最快的手工模拟客户端工具服务端启动后直接在终端执行telnet 127.0.0.1 8080就能手动输入数据测试连接。第二tcpdump或Wireshark用来抓包看三次握手、数据段交互排查TCP层面的问题。第三纯代码层面的日志这是我自己最常用的每个连接建立、断开、收发数据都打印关键信息。刚开始写服务端时建议三步都用上。telnet验证“能不能连上”日志验证“代码里走到哪一步了”抓包验证“网络层到底发生了什么”。这三个工具配合起来基本没有排查不了的问题。3. TCP开发的核心细节3.1 三次握手、listen backlog与端口复用先解决一个基础问题TCP三次握手到底是谁做的答案是内核。你在Python里调用connect()时内核会自动完成SYN、SYN-ACK、ACK这三个包的交换你的代码根本感知不到这一步。但有一个参数你需要关注就是listen()里的backlog。它表示内核为这个监听套接字维护的“已完成连接队列”的最大长度。简单理解就是客户端连接进来后如果服务端还没调用accept()取走这个连接会暂时排队。backlog设置太小高峰期就会出现客户端connect超时。再看一个隐蔽的坑服务端重启时如果报Address already in use多半是之前的连接处于TIME_WAIT状态端口还没释放。解法很简单在bind之前加上这一行server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)这个设置允许端口复用的时机提前让服务端可以快速重启。我写任何服务端代码都会无脑加上这行已经成为肌肉记忆了。3.2 粘包问题字节流没有边界这是TCP开发里最容易卡住新手的一道坎而且面试必问。TCP本质上是字节流协议它只保证字节按顺序到达不保证“一次send对应一次recv”。也就是说你连续调用两次sendall()发送两条小消息对方一次recv()可能就把两条消息一起读走了这就是粘包。网上有人通过time.sleep()来避免粘包那完全是撞运气并发一高照样崩。正规做法是自定义消息边界我常用的是“固定长度消息头 消息体”的方案。消息头4个字节用struct.pack(!I, length)打包成网络字节序消息体放实际数据。import struct def send_msg(sock, data: bytes): # 先把消息长度打包成4字节再拼接消息体一起发送 header struct.pack(!I, len(data)) sock.sendall(header data) def recv_exact(sock, count: int) - bytes: # 循环接收保证收满count个字节 buffer b while len(buffer) count: chunk sock.recv(count - len(buffer)) if not chunk: raise ConnectionError(connection closed) buffer chunk return buffer def recv_msg(sock) - bytes: header recv_exact(sock, 4) length struct.unpack(!I, header)[0] return recv_exact(sock, length)这里recv_exact()是关键因为recv(4)也可能只收到2个字节必须循环收满。很多人只写一次recv(4)数据一多就会出现解析错位。这个小函数我在所有TCP项目里都会复用。3.3 优雅关闭与半关闭先分清两个方法close()和shutdown()。close()是释放文件描述符引用计数归零后连接才会真正关闭shutdown()则是直接切断数据收发通道可以只关发送、只关接收或者全关。业务上有一种场景服务端发送完响应后并不想等客户端再发数据只想单方面关闭发送通道同时还能继续接收。这时候用shutdown(SHUT_WR)实现半关闭就非常合适。如果不做半关闭双方可能都在傻等对方先关形成死等状态。一个简单的原则短连接场景里服务端在发送完响应后调用shutdown(socket.SHUT_WR)告诉客户端“数据发完了”客户端收到EOF关闭连接这个模式在自研TCP协议时特别实用。4. 单进程TCP服务端先把一条链路跑通4.1 最小可用的单进程服务端不急着上多进程先写一个能跑的服务端把accept、recv、send这三个动作练扎实。下面这段代码逻辑很简单监听到一个客户端连接后进入循环接收数据再把数据原样返回也就是一个echo server。import socket server socket.socket(socket.AF_INET, socket.SOCK_STREAM) server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server.bind((0.0.0.0, 8080)) server.listen(5) print(server listening on 8080) while True: conn, addr server.accept() print(fclient connected: {addr}) try: while True: data conn.recv(1024) if not data: break conn.sendall(data) except ConnectionResetError as e: print(fclient {addr} reset: {e}) finally: conn.close()这个代码能跑通但有一个非常明显的问题accept()和recv()都是阻塞调用如果第一个客户端连接上来后一直不发送数据服务端就会卡在recv()第二个客户端的accept()根本执行不到。也就是说这个服务端同一时间只能服务一个客户端。4.2 单进程模型的瓶颈在哪里单进程的瓶颈本质上是“阻塞”二字。网络服务的特点是大部分时间都在等I/OCPU空闲得很但阻塞模型把进程卡死了后续连接全部堵在门外。这个阶段有人会想给每个连接开一个线程是不是就解决了答案是可以但要付出额外代价。Python线程受GIL影响虽然网络I/O等待时GIL会释放多线程确实能提高并发能力可线程的创建、切换、销毁都有开销。更重要的是线程之间共享所有内存一个线程写坏全局状态整个进程都跟着遭殃。所以我的建议是如果你想做短连接、请求响应模型的服务端多进程通常比多线程更省心这也是我下面重点展开的内容。但你要理解一件事多进程也好多线程也罢它们解决的是“如何同时服务多个连接”的问题而不是“如何让一个CPU跑得更快”的问题。5. 多进程从串行到并发的关键一步5.1 多进程和多线程到底怎么选这是一个被问了无数次的问题我把我实际选型时的判断标准写出来。多线程的优势是共享内存方便线程之间可以直接读写同一个变量配合锁就能协同工作劣势是GIL让CPU密集型任务无法利用多核而且一个线程崩了可能拖垮整个进程。多进程的优势是每个进程有独立的GIL能真正并行利用多核CPU进程崩溃不会影响其他子进程隔离性好劣势是进程间通信繁琐创建成本比线程高。放到TCP服务端这个场景里主要工作就是收发数据和协议解析属于I/O密集型加少量计算。多进程模型在稳定性和隔离性上明显更好所以生产环境大批量使用pre-fork模型也就是下文的master-worker模式。多线程模型当然也能用但如果你刚开始做网络编程我建议从多进程入手因为内存隔离能让你少踩很多“共享状态被意外修改”的坑。5.2 用multiprocessing.Process实现pre-fork模型多进程TCP服务端有两个主流模型一个是“主进程只accept然后把连接交给子进程处理”另一个是“主进程listen多个子进程同时accept”。前者逻辑清晰但需要额外的进程间通信来传递socket后者就是经典的pre-fork代码更简单性能也不差。pre-fork的核心是父进程创建监听套接字后通过multiprocessing.Process派生出多个子进程子进程继承父进程的文件描述符然后各自阻塞在accept()上。当一个新连接到达时内核会唤醒其中一个进程来处理。import socket import multiprocessing def worker(server): while True: conn, addr server.accept() print(fsubprocess({multiprocessing.current_process().name}) fhandle client: {addr}) try: while True: data conn.recv(1024) if not data: break conn.sendall(data) except ConnectionResetError: pass finally: 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, 8080)) server.listen(128) workers [] for _ in range(multiprocessing.cpu_count()): p multiprocessing.Process(targetworker, args(server,)) p.daemon True p.start() workers.append(p) for p in workers: p.join() if __name__ __main__: main()这里有几个细节需要注意。daemonTrue保证主进程退出时子进程也一起退出否则会出现僵尸进程或孤儿进程。worker函数里使用socket..accept()返回值conn这个conn是从内核拿到的新文件描述符多个子进程之间是各自独立的不会互相干扰。5.3 多进程下的临界资源日志与计数器用上多进程之后你马上会踩到另一个坑多个子进程同时往同一个文件里写日志会产生乱行多个子进程同时修改同一个计数器计数会丢。原因是每个子进程都有独立的内存空间普通的全局变量只在单个进程内生效。解决思路有两条。如果只是统计连接次数可以用multiprocessing.Value配合加锁import multiprocessing counter multiprocessing.Value(i, 0) lock multiprocessing.Lock() def handle_conn(): with lock: counter.value 1如果需要写日志更推荐的办法是用multiprocessing.Queue子进程把日志消息放进队列主进程单独开一个线程负责写文件。这样日志顺序可控写入压力也小。这个模式在下面的实操代码里我会写一个简化版本。6. 实战拆解可扩展的多进程TCP服务端6.1 完整代码把前面那些知识点合在一起我给出一个相对完整的版本它具备这些能力pre-fork多进程并发、连接数统计、日志Queue收集、优雅处理连接异常。代码不长但每一行都有实际用途。import socket import struct import multiprocessing import logging import time def recv_exact(conn, count): buf b while len(buf) count: chunk conn.recv(count - len(buf)) if not chunk: raise ConnectionError(client closed) buf chunk return buf def recv_msg(conn): header recv_exact(conn, 4) length struct.unpack(!I, header)[0] if length 1024 * 1024: raise ValueError(message too large) return recv_exact(conn, length) def worker(server, log_queue, counter, lock): while True: conn, addr server.accept() with lock: counter.value 1 log_queue.put(f[{time.strftime(%H:%M:%S)}] fconn{counter.value} from{addr}) try: while True: data recv_msg(conn) ack struct.pack(!I, len(data)) data conn.sendall(ack) except (ConnectionError, ValueError, socket.timeout): pass finally: conn.close() log_queue.put(f[{time.strftime(%H:%M:%S)}] fclose from{addr}) def log_listener(queue): logging.basicConfig( filenameserver.log, levellogging.INFO, format%(asctime)s %(message)s ) while True: msg queue.get() logging.info(msg) 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, 8080)) server.listen(256) log_queue multiprocessing.Queue() counter multiprocessing.Value(i, 0) lock multiprocessing.Lock() logger_proc multiprocessing.Process(targetlog_listener, args(log_queue,)) logger_proc.daemon True logger_proc.start() workers [] for _ in range(multiprocessing.cpu_count()): p multiprocessing.Process(targetworker, args(server, log_queue, counter, lock)) p.daemon True p.start() workers.append(p) for p in workers: p.join() if __name__ __main__: main()代码里我顺手加上了消息长度校验超过1MB直接丢弃。这个保护在生产环境非常重要否则恶意客户端可以发送一个超大长度字段让服务端一直循环等待收满数据白白占着连接。6.2 关键配置和参数解读这里逐个聊一聊参数的选择理由。进程数multiprocessing.cpu_count()我一般取CPU物理核心数或者稍微多一点比如1.5倍。网络服务是I/O密集型进程数比核心数略高通常没问题但你需要在内存和CPU之间找平衡。每个子进程都有一份独立的Python解释器内存占用会成倍增长盲堆进程数只会让调度开销变大。backlog写256是预留一定余量。如果这个服务端面向公网连接瞬时爆发量高建议调到1024以上如果只做内网小流量服务128就够。这个值不必刻意拉满因为还有内核参数net.core.somaxconn在限制上限设置再高也没用。daemonTrue父进程退出时子进程会被强制回收。如果业务上需要子进程在父进程退出后继续运行比如做守护进程那就别设daemon但那样你就得自己处理SIGTERM信号复杂度会高很多。6.3 多进程下的数据隔离与通信我在worker里用multiprocessing.Value保存计数器用multiprocessing.Queue传日志。这是多进程通信最常用的两种方式。Value在底层是共享内存配合Lock使用能保证原子性Queue底层是管道加锁多个进程往里放数据时序列化后写入顺序基本有保障。有一点必须说清楚子进程之间不是完全隔离的它们共享父进程打开的文件描述符这就是为什么所有worker都能调用server.accept()。但在Python的对象层面上每个子进程都有自己的内存视图你随便给一个worker设置全局变量其他worker完全看不到所以别指望“全局变量跨进程共享”。7. 压测验证与性能对比7.1 本地压测脚本怎么写写网络服务不压测等于白写数据最能说明问题。我常用的压测思路是模拟并发客户端每个客户端连续发送100条消息统计总耗时和成功率。为了减少网络因素干扰测试走本地回环地址。import socket import struct import time from concurrent.futures import ThreadPoolExecutor def client(index): sock socket.create_connection((127.0.0.1, 8080)) for i in range(100): data fmsg-{index}-{i}.encode() header struct.pack(!I, len(data)) sock.sendall(header data) resp_header recv_exact(sock, 4) resp_length struct.unpack(!I, resp_header)[0] recv_exact(sock, resp_length) sock.close() def recv_exact(sock, n): buf b while len(buf) n: chunk sock.recv(n - len(buf)) if not chunk: break buf chunk return buf start time.time() with ThreadPoolExecutor(max_workers50) as pool: list(pool.map(client, range(50))) print(fcost: {time.time() - start:.2f}s)这个压测脚本虽然简单但能说明并发能力。我建议你也分别跑一下单进程、多线程、多进程三个版本用同样的压测数据做对比。7.2 实测数据我本机是8核心16线程的机器三种模式的粗略结果如下模型并发客户端数每条连接消息数总耗时单进程阻塞501005.3s多线程8线程501001.9s多进程8进程501001.6s单进程最慢的原因是串行阻塞连接全部排队等待多线程和多进程能并发处理耗时会明显下降。多进程比多线程快一点点但差距没有想象中那么大毕竟这里的核心操作是I/OPython线程在recv等待时同样会释放GIL。多进程的真正优势在大流量和高稳定要求场景下更明显比如单个客户端连接数上万时线程切换成本和GIL争用会成为瓶颈此时多进程的心跳机制和隔离性就更重要。8. 常见问题与排查技巧实录8.1 高频报错速查表报错信息原因解决办法Address already in use端口被占用或处于TIME_WAITbind前设置SO_REUSEADDRBrokenPipeError对端已关闭本端还在send捕获异常关闭连接ConnectionResetError对端重置连接检查协议或超时设置OSError: [Errno 24] Too many open files文件描述符耗尽增大ulimit -n或减少长连接数TypeError: a bytes-like object is requiredsend时传了str而非bytes统一用encode()转为bytesConnectionResetError是新手最容易遇到又最容易忽略的它通常不是代码逻辑问题而是对端直接发送了RST包。RST的触发原因很多比如客户端进程崩溃、主动关闭未完成的数据传输、防火墙干预等。你只要保证服务端代码里对每个recv()、send()都做好异常捕获这个报错基本不会拖垮进程。8.2 实战心得与排查习惯最后一个部分我分享几个长期以来帮自己省时间的小习惯。第一个习惯是给服务端增加一个独立的调试出口比如加个--debug参数打印每个连接的五元组信息源IP、源端口、目标IP、目标端口、协议。高并发下想定位某条异常连接时没有这些信息根本无从下手。第二个习惯是不要直接在子进程里写文件日志尤其是多进程场景。你以为多个进程写同一个文件没多大事实际会出现行交错、日志丢失等问题。我后来一律改用Queue 单进程日志器问题直接消失。第三个习惯是定期做容量估算。每次上线前我会根据请求量估算需要多少并发进程、多少个文件描述符。公式很简单如果每个连接占用两个文件描述符socket epoll那1万个连接需要约2万个文件描述符系统默认1024肯定不够需要提前调整ulimit -n。第四个习惯是关注TIME_WAIT和CLOSE_WAIT状态。用ss -ant查看连接状态如果大量连接卡在CLOSE_WAIT说明服务端没有正确关闭连接多半是代码里漏了close()。如果TIME_WAIT过多通常说明服务端主动断开了大量连接短连接场景下这是正常现象结合SO_REUSEADDR处理即可。说到这儿回头再看“Socket套接字、TCP开发、多进程”这一章你会发现它们其实是一件事用底层TCP协议做可靠网络通信用多进程把单机并发能力榨干。我自己在做网络编程的第一年基本就是把这几段代码反复改写、压测、对比才真正建立起对服务端模型的直觉。你把这篇文章里的代码跑通之后下一步可以试着加一个简单的协议解析比如把echo server改成支持JSON消息这样离一个真实业务服务端就更近了。