Python网络编程实战：Socket套接字+TCP开发+多进程并发服务端

如果你已经走完了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表示IPv4，SOCK_STREAM表示TCP client = socket.socket(socket.AF_INET, socket.SOCK_STREAM) # 发起TCP连接，其实就是完成三次握手 client.connect(("127.0.0.1", 8080)) # 发送数据，sendall会等数据都发完才返回 client.sendall(b"GET / 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(f"client connected: {addr}") try: while True: data = conn.recv(1024) if not data: break conn.sendall(data) except ConnectionResetError as e: print(f"client {addr} reset: {e}") finally: conn.close()

这个代码能跑通，但有一个非常明显的问题：accept()和recv()都是阻塞调用，如果第一个客户端连接上来后一直不发送数据，服务端就会卡在recv()，第二个客户端的accept()根本执行不到。也就是说，这个服务端同一时间只能服务一个客户端。

4.2 单进程模型的瓶颈在哪里

单进程的瓶颈本质上是“阻塞”二字。网络服务的特点是大部分时间都在等I/O，CPU空闲得很，但阻塞模型把进程卡死了，后续连接全部堵在门外。

这个阶段有人会想，给每个连接开一个线程是不是就解决了？答案是可以，但要付出额外代价。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(f"subprocess({multiprocessing.current_process().name}) " f"handle 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(target=worker, args=(server,)) p.daemon = True p.start() workers.append(p) for p in workers: p.join() if __name__ == "__main__": main()

这里有几个细节需要注意。daemon=True保证主进程退出时子进程也一起退出，否则会出现僵尸进程或孤儿进程。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')}] " f"conn={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')}] " f"close from={addr}") def log_listener(queue): logging.basicConfig( filename="server.log", level=logging.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(target=log_listener, args=(log_queue,)) logger_proc.daemon = True logger_proc.start() workers = [] for _ in range(multiprocessing.cpu_count()): p = multiprocessing.Process(target=worker, 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在限制上限，设置再高也没用。

daemon=True：父进程退出时子进程会被强制回收。如果业务上需要子进程在父进程退出后继续运行，比如做守护进程，那就别设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 = f"msg-{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_workers=50) as pool: list(pool.map(client, range(50))) print(f"cost: {time.time() - start:.2f}s")

这个压测脚本虽然简单，但能说明并发能力。我建议你也分别跑一下单进程、多线程、多进程三个版本，用同样的压测数据做对比。

7.2 实测数据

我本机是8核心16线程的机器，三种模式的粗略结果如下：

单进程最慢的原因是串行阻塞，连接全部排队等待；多线程和多进程能并发处理，耗时会明显下降。多进程比多线程快一点点，但差距没有想象中那么大，毕竟这里的核心操作是I/O，Python线程在recv等待时同样会释放GIL。多进程的真正优势在大流量和高稳定要求场景下更明显，比如单个客户端连接数上万时，线程切换成本和GIL争用会成为瓶颈，此时多进程的心跳机制和隔离性就更重要。

8. 常见问题与排查技巧实录

8.1 高频报错速查表

ConnectionResetError是新手最容易遇到又最容易忽略的，它通常不是代码逻辑问题，而是对端直接发送了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消息，这样离一个真实业务服务端就更近了。