Python网络编程从TCP/IP到依赖管理：socket与requirements.txt工程化实践

1. 为什么能跑就行的项目往往倒在环境搭建这一步先说个我这些年的真实感受很多人学 Python 网络编程第一关根本不是 TCP 三次握手也不是 socket 各种坑而是环境本身就一塌糊涂。我见过不少同事的笔记本Python 装了三四个版本路径乱成一锅粥项目里依赖靠 pip 一个一个手动装装到一半报错就百度百度完又往全局环境里塞。最后项目换台电脑就跑不起来交出去直接被运维打回。这类问题在涉及 TCP/IP 网络编程的项目里尤其致命——因为网络程序调试本身就依赖多端配合、端口联通、防火墙规则这些外部条件一旦环境不稳定你根本分不清是代码 bug 还是依赖缺失排查成本直接翻倍。所以这篇文章我不想一上来就写 socket 语法。我想把两个看似不相关、实际上紧密耦合的主题放在一起讲清楚TCP/IP 网络编程的核心链路以及 requirements.txt 的工程化最佳实践。为什么放一起因为网络编程练手阶段最常用的就是本机开服务端、再开客户端自测这天然就是一个多模块、多依赖的项目场景。依赖管理做不好你连一个稳定的回环测试环境都搭不起来。反过来说光会写 socket 代码却不懂依赖管理你写出来的东西也走不出自己这台电脑。先给不同基础的读者定位一下如果你刚学 Python 语法这篇文章能帮你理清 socket 编程的主干路径也能告诉你 requirements.txt 到底解决什么问题如果你已经写过一些网络脚本本文重点在工程化那部分——多环境一致性、依赖锁定、兼容性边界这些才是让你从自己跑得通走向别人也跑得通的关键。2. TCP/IP 协议栈写代码前必须建立的最小认知框架很多教程一上来就让你import socket然后socket.socket()看似简单但一旦遇到半关闭、粘包、连接重置这些问题没有协议栈层面的认知你只能靠试错碰运气。所以我想先花一小节把写 Python socket 代码真正需要的那部分 TCP/IP 底子讲清楚不讲教科书上那些冗长的分层细节只讲和代码行为直接相关的映射关系。2.1 四层模型和你写的每一行代码之间的关系TCP/IP 模型通常分成四层应用层、传输层、网络层、网络接口层。在 Python socket 编程里你平时打交道的主要是前两层但网络层直接决定了你填的 IP 地址是什么这层理解偏了后面全是玄学问题。应用层HTTP、SSH、你自己用 socket 封装的协议都属于这一层。Python 里你操作的就是这一层的数据——你 send 出去的字节、recv 回来的字节。传输层TCP 和 UDP 在这里。TCP 提供可靠字节流保证数据不丢不乱UDP 只提供报文传输不管顺序也不管丢包所以性能开销小很多。socket(AF_INET, SOCK_STREAM)选 TCPSOCK_DGRAM选 UDP。网络层IP 协议在这里。你写程序时填的127.0.0.1、192.168.1.10就是三层地址。它负责把数据从一台机器路由到另一台机器TCP 报文只是它负载里的一段。网络接口层网卡驱动、以太网帧这些。平时写应用层代码基本感知不到但MTU最大传输单元、Wi-Fi 信号差导致的丢包都发生在这附近。那为什么说这个框架重要因为你在代码里遇到的很多怪现象本质上都能映射到某一层。比如你发现自己 send 成功但服务端迟迟收不到完整数据先别怀疑代码乱序很可能是 TCP 分段和粘包问题——这是传输层对字节流的处理方式决定的。又比如你连远程服务器总超时但 IP 能 ping 通这时候你要想的是传输层端口是否可达而不是一遍遍检查应用层逻辑。2.2 握手、挥手、状态迁移服务端和客户端的代码差异都源于这里TCP 三次握手的过程大家多少听过客户端发 SYN服务端回 SYNACK客户端再回 ACK连接建立。这不仅是理论它直接决定了 socket API 的阻塞行为。服务端代码里listen()启动了监听操作系统在后台维护一个半连接队列SYN 队列和一个全连接队列accept 队列。三次握手完成后连接对象会被放进全连接队列你的accept()只是从这个队列里取。这就是为什么accept()在没有连接时会被阻塞住——不是程序卡死了是在等队列里有货。四次挥手也有直接体现。主动关闭方发送 FIN然后进入 FIN_WAIT_2被动关闭方进入 CLOSE_WAIT。如果你在代码里只关了服务端、没管客户端的状态很容易出现端口资源不释放的情况。写网络程序最常用的排障命令netstat -an | grep 端口里能看到一大堆 TIME_WAIT 和 CLOSE_WAIT不了解状态迁移你看到 TIME_WAIT 就会慌以为出 bug 了。当然代码层面有一个更直观的问题close()和shutdown()的区别。close()是彻底关闭 socket释放文件描述符shutdown()可以只关闭发送方向或接收方向。TCP 是双工的被动关闭方如果只close()不shutdown()在某些场景下会导致 FIN 的时序不符合预期。这块后面讲代码时我会用一个例子说明。提示如果你发现自己写的客户端发完数据立即close()服务端recv()立刻返回空字节串这其实是 TCP FIN 正常传递的结果——EOF 信号被映射成了 recv 返回 0。这不是错误而是一个常用设计技巧后面讲如何优雅关连接时会用到。2.3 别混淆 IP 层和 TCP 层connect 超时与 ping 不通是两码事我遇到很多新手排查网络问题时第一反应是ping对端 IP。ping 走的是 ICMP属于网络层工具不涉及端口。而你的程序用的是 TCP它对端口的关注远大于对 IP 是否可达的关注。可能出现的情况是ping 得通业务连不上因为端口被防火墙挡了也可能反过来ping 不通对端禁 ping但业务端口照样能连。这套区分很重要。排查思路应该顺着分层走先确认网络层通不通——用ping或tracerouteWindows 下是tracert。确认 TCP 层通不通——用telnet 192.168.1.10 8080能连上说明 TCP 握手成功端口是开放的。再回到应用层看数据格式对不对——这才是你的 Python 代码重点关注的地方。如果第 1 步都不通你调 socket 代码是白费力气。很多新手在 Windows 本机测试时明明端口没问题程序却连不上排查半天发现是 Windows 防火墙拦了 Python 进程。这种问题不属于 TCP/IP 协议逻辑但属于你没把协议栈模型和实施部署结合起来考虑属于应用层以上的人类操作失误。3. 实操 TCP 编程从最基本的一对一通信开始我尽量用可复制、可修改的最小案例来讲每一段代码后面都标注为什么要这么设计。代码环境默认 Python 3.10Linux 和 Windows 均可用只需要把运行窗口分成两个终端。3.1 服务端监听与客户端连接的最小骨架先来一个最基础的回声服务echo server。它做三件事监听端口、接收连接、把收到的原样发回去。这个场景能把 socket 的核心流程完整走一遍。# server_echo.py import socket def main(): host 127.0.0.1 # 回环地址只接受本机连接适合自测 port 8080 server_sock socket.socket(socket.AF_INET, socket.SOCK_STREAM) server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) # 见下文解释 server_sock.bind((host, port)) server_sock.listen(5) print(f监听 {host}:{port}) while True: conn, addr server_sock.accept() print(f客户端已连接来自 {addr}) data conn.recv(1024) while data: conn.sendall(data) # sendall 和 send 的区别见下文 data conn.recv(1024) conn.close() print(f客户端 {addr} 已断开) if __name__ __main__: main()# client_echo.py import socket def main(): host 127.0.0.1 port 8080 client_sock socket.socket(socket.AF_INET, socket.SOCK_STREAM) client_sock.connect((host, port)) print(已连接到服务端) try: text Hello TCP/IP client_sock.sendall(text.encode(utf-8)) response client_sock.recv(1024) print(f收到回显: {response.decode(utf-8)}) finally: client_sock.close() if __name__ __main__: main()这两个文件加起来不到四十行但涵盖了整个 TCP 生命周期的核心 API。先跑服务端再跑客户端你会看到标准输出里服务端打印了连接和断开的信息。为什么加setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)如果你调试时敲了 CtrlC 强杀服务端程序操作系统不会立刻释放 8080 端口默认要等 TIME_WAIT 状态过去。不设这个选项的话你重跑服务端大概率会报OSError: [Errno 98] Address already in use。设了之后端口在重启场景下的复用会顺畅得多。为什么用sendall而不是send这是 TCP 一个著名的坑send并不能保证把你给的字节全部推入发送缓冲区它可能只发送一部分就返回了返回实际发送的字节数。你必须手动检查返回值循环发送剩余部分。而sendall内部就是这个逻辑的封装要么全部发送成功要么抛异常省掉了你自己写 while 循环的麻烦。很多人刚接触时意识不到这一点直接用send传输大文件结果文件莫名少一段还以为是网络丢包。3.2 粘包、半包的根因与三种解法粘包可能是 Python 网络编程被问得最多的问题。它的现象是客户端连续两次 send服务端 recv 一次就把两段数据都读回来了或者反过来一次 send 的数据被分成了两次 recv 才取全。先说清楚为什么会这样。TCP 是字节流协议它不关心你上层怎么切分数据——你 send 了两次在 TCP 层可能被合并进同一个 TCP 报文段也可能被拆成多个报文段。这由 TCP 的滑动窗口、Nagle 算法、系统缓冲区共同决定。应用层能看到的就是一条没有边界的字节流。所以严格来说TCP 根本没有消息概念只有字节流。粘包、半包是应用层对消息边界感知缺失导致的。要解决这个问题本质是给字节流加上帧的概念让接收方能判断一条消息从哪开始到哪里结束。常用方案有这三种方案一固定长度帧每条消息固定 N 字节不足补零。实现最简单但浪费带宽且每条消息长度差异大的场景根本不适合。适合协议非常简单、长度基本固定的场景。方案二分隔符帧消息结尾放特殊字符比如\n常用于文本协议接收方按分隔符切割。HTTP/1.1 的消息头就是这么干的以空行结束。问题是消息本身不能包含该分隔符需要转义处理。方案三长度前缀最推荐发送每条消息前先用固定字节数比如 4 字节声明后续长度然后接收方先读长度再按长度读对等字节。这是二进制协议最常用的方案比如很多物联网设备协议、Python 的 struct 打包方式都基于此。# server_length.py 使用 struct 打包长度头 import socket import struct def recv_exact(conn, n): chunks [] remaining n while remaining 0: chunk conn.recv(remaining) if not chunk: # 连接关闭但数据未取够属于异常情况 raise ConnectionError(连接提前关闭数据不完整) chunks.append(chunk) remaining - len(chunk) return b.join(chunks) def recv_message(conn): header recv_exact(conn, 4) (msg_len,) struct.unpack(!I, header) # 网络字节序大端 return recv_exact(conn, msg_len) def main(): server_sock socket.socket(socket.AF_INET, socket.SOCK_STREAM) server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server_sock.bind((127.0.0.1, 8080)) server_sock.listen(5) print(长度前缀协议服务端已启动) while True: conn, addr server_sock.accept() print(f客户端连接: {addr}) while True: try: msg recv_message(conn) print(f收到完整消息长度 {len(msg)}: {msg!r}) except ConnectionError: print(f{addr} 连接关闭) break conn.close() if __name__ __main__: main()这里我实现了一个recv_exact专门用来精确接收 N 字节。这是解决半包的核心工具因为recv(n)并不保证一次就返回 n 字节网络拥塞、缓冲区分配都可能导致返回的字节数少于请求数。recv_exact内部用循环把数据拼齐直到目标长度满足或用完连接。客户端对应发送时把长度头和数据一起发import socket import struct def send_message(sock, data: bytes): header struct.pack(!I, len(data)) sock.sendall(header data) sock socket.socket(socket.AF_INET, socket.SOCK_STREAM) sock.connect((127.0.0.1, 8080)) send_message(sock, bhello) send_message(sock, bworld) sock.close()这就是标准的长度前缀 载体设计。注意我用了!I感叹号代表网络字节序大端这是 TCP/IP 协议栈的标准字节序要求保证跨平台解码一致。如果你用小端的两个程序跑在不同架构的机器上长度字段就会解析颠倒这也是跨端兼容性里容易忽视的小细节。3.3 处理并发连接多线程、selectors 与 asyncio 的取舍上面那个服务端是串行处理的同一时间只能服务一个客户端。第二个客户端连上来得等第一个客户端断开才能被 accept。真实场景显然不能这样所以并发处理是必须考虑的。Python 里做并发 socket 服务端常见思路有三条多线程线程 per 连接最简单直观主循环 accept拿到新连接就开一个线程去处理。适合连接数不多几十到几百的场景代码清晰。selectors/IO 多路复用用单线程监听多个 socket 的读写事件连接数很多上千时比线程省资源。我在生产环境维护过的边缘采集服务就用的这类模型稳是稳但对新手来说回调式写法可读性差。asyncio现代 Python 异步框架协程写法更接近同步代码同时路径上不阻塞线程。适合强 IO 密集型任务但如果你对整个异步事件循环模型不熟排错体验会很难受。我给一个小型多线程服务端模板因为它在教学和中小业务里平衡性最好# server_threaded.py import socket import threading def handle_client(conn, addr): print(f处理客户端 {addr} 的线程: {threading.current_thread().name}) try: while True: data conn.recv(1024) if not data: break conn.sendall(data.upper()) # 将小写转为大写回显 finally: conn.close() print(f客户端 {addr} 连接关闭线程退出) 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(5) print(多线程服务端启动) while True: conn, addr server.accept() t threading.Thread(targethandle_client, args(conn, addr), daemonTrue) t.start() if __name__ __main__: main()这里有个值得说的点conn在 accept 时只有一个引用交给线程后主线程不应该再去关闭它否则可能出现线程还在读写文件描述符已经被主线程释放的情况。所以handle_client里通过try/finally管理连接生命周期主线程只管接收新连接不碰已有连接的生死。如果用 asyncio 写等价版本核心代码是import asyncio async def handle_client(reader, writer): addr writer.get_extra_info(peername) print(f客户端连接: {addr}) while data : await reader.read(1024): writer.write(data.upper()) await writer.drain() writer.close() await writer.wait_closed() async def main(): server await asyncio.start_server(handle_client, 0.0.0.0, 8080) async with server: await server.serve_forever() if __name__ __main__: asyncio.run(main())asyncio 版本在处理高并发、尤其是海量长连接时比多线程模型省大量内存和上下文切换开销。新手如果暂时不了解异步也可以先把多线程版本吃透理解了阻塞 IO 与线程调度的关系再进入 asyncio 会顺畅很多。4. requirements.txt 的工程化梳理从能跑到可复现有人可能觉得奇怪一篇讲 TCP 网络编程的文章为什么花整章讲依赖管理。我的理由是你不可能总在干净环境里跑网络程序。你会遇到部署到服务器要装依赖、接手别人代码要恢复环境、升级 Python 版本后库不兼容等问题。没有规范的依赖清单这些环节步步是坑。4.1 为什么不能只用pip freeze requirements.txt?这是最常见的直觉做法却不推荐直接使用。原因有三第一pip freeze会把当前环境中所有包都列出来包括那些不是当前项目直接依赖的包比如你为了写爬虫测试装的库也列进去了。项目本身可能只需要七八个包冻结出来却有三五十行后续维护者看着一头雾水。第二pip freeze会列出所有传递依赖间接依赖。不是说传递依赖不该记录而是全部平铺记录会造成哪些是直接依赖、哪些是间接依赖完全无法区分。你在 requirements.txt 里只需要声明直接依赖让 pip 自己解析间接依赖即可这样升级直接依赖版本时间接依赖的解析更灵活不容易出现互相锁死。第三同一个环境被多项目复用时pip freeze会把其他项目的依赖也混进来输出结果不稳定。正确做法是按需手动维护 requirements.txt只写直接依赖并给出合理的版本范围。类似 .NET 的 packages.config 或 Go 的 go.mod 定位requirements.txt 是项目入口清单不应该是环境快照。那什么时候需要锁定版本打包发布、线上回滚需要可重复安装时应该用pip freeze requirements.lock生成一份完全锁定的清单但这份锁文件与 requirements.txt 分开管理前者面向部署复现后者面向源码维护。4.2 基础语法、版本范围与传递依赖requirements.txt 本质是 pip 安装指令的集合。每行一个包支持多种格式requests2.31.0 numpy1.26,2.0 pandas~2.1.0 socket # 注意这个看到最后这行我先提醒一下socket模块是 Python 标准库自带的绝不能写进 requirements.txt。因为 pip 去 PyPI 找socket很可能找到一个同名但完全不相关的第三方包安装后不但没用还会污染环境。还有sys、os、threading等同样不要装。版本描述符解释2.31.0精确锁定部署时常见。缺点是过严间接依赖之间可能产生冲突。1.26,2.0指定上下限推荐用于主依赖。~2.1.0等价于2.1.0,2.1.*意思是 2.1.x 的补丁版本都可禁止跳到 2.2。*任意版本不推荐直接裸写等于让维护者无法判断当前跑的是哪个版本。挑版本号的原则我有三条主版本不要随便跨跨大版本往往意味着 API 断裂。次版本建议锁定最低值且留出补丁版本空间类似1.26,2.0。遇到文档明确警告不兼容组合的写明注释例如# 不要降到 2.0 以下否则 xx 接口失效。4.3 多环境管理使用 pip-tools 或显式环境拆分如果你的项目需要同时支持开发环境和生产环境最简单的是拆成多份文件requirements-base.txt两边都需要的核心依赖。requirements-dev.txt包含 pytest、debug 工具等第一行写-r requirements-base.txt表示继承。requirements-prod.txt只装生产环境依赖也可以继承 base。这种继承写法 pip 完全支持而且语义清晰# requirements-dev.txt -r requirements-base.txt pytest8.0,9.0安装时执行pip install -r requirements-dev.txt。这个方案比装一堆管理工具更通用团队协作时也容易 review。如果想更进一步可以使用pip-tools它的思路是先写requirements.in只写直接依赖然后用pip-compile自动生成完整的requirements.txt间接依赖和版本锁定全部自动处理。做法是pip install pip-tools # 写入 requirements.in例如 requests2.31 pip-compile requirements.in -o requirements.txt在部署时用pip install -r requirements.txt安装能保证每台机器都装出完全相同的依赖树。4.4 在 Windows、Linux 混用时规避二进制依赖的坑网络编程项目用到的一些库比如pyserial、psutil、cryptography或者某些绑定了 C 扩展的库在 Windows 和 Linux 上安装机制完全不同。Windows 上很多时候有官方预编译的 wheel直接装上即可Linux 某些精简发行版可能缺少编译工具链pip 会尝试从源码编译这时需要gcc、python3-dev等系统包否则报一堆缺失头文件和Command python setup.py egg_info failed。处理办法优先选同时提供多平台 wheel 的包可以从 PyPI 页面确认。如果 Linux 上编译报错检查是否装了build-essential、python3-dev。在云服务器上常因为少了这些导致依赖装不上。对 pyserial 这类串口相关库如果在 Windows 下遇预编译 wheel 缺失考虑改用pyserial的官方文档推荐的编译方式或者换用纯 Python 实现的替代方案。不要跨平台共享同一个 venv 目录Windows 生成的文件和 Linux 不兼容直接在目标平台上重新建虚拟环境。关于虚拟环境我多说一句现代 Python 有内置venv模块命令就三条不要装一堆额外包python -m venv .venv # Windows 激活 .venv\Scripts\activate # Linux/macOS 激活 source .venv/bin/activate项目提交代码时把.venv加进.gitignore不要把虚拟环境目录提交到仓库。requirements.txt 才是团队协作的环境载体。5. 依赖管理是不是一定需要pip install -r requirements.txt才能用说一个我在观看 Windows 和 Linux 双向合作项目时的高频问题明明在自己机器上pip install -r requirements.txt装好了一换机器还是错误频出。其实问题通常不在命令本身而在于你安装时的平台和本次运行时的平台不一致。依赖分为纯 Python 包和带二进制扩展的包。纯 Python 包跨平台相对安全带 C 扩展的包如cryptography、bcrypt、pymssql在 Windows 上预编译的 wheel 跟 Linux 上从源码编译的产物不是同一个东西。你在 Windows 里pip freeze锁定了某个版本到 Linux 服务器上重新安装pip 会重新选择对应平台 wheel只要版本范围没写死多半能装上真正把范围写死之后反而可能出现该版本在 Linux 上没有对应 wheel 的情况从而退回到源码编译路线一连串缺依赖就来了。所以实践上requirements.txt 里的版本范围要宽一点不要每条都精确版本。位数精确留给 CI/CD 的锁文件手工维护的 requirements.txt 保持灵活。这样在不同平台、不同 Python 补丁版本间才能装得上、跑得起。6. 用本机双进程实战演练 TCP 传输并顺手验证依赖管理网络编程和其他方向有一个很大的不同它几乎天生适合自己给自己当服务端和客户端。下面这套演练我经常推荐给新同事它能把协议栈、socket API 和依赖管理一次性串起来零云服务器成本仅依赖 Python 标准库。6.1 搭建一个本机测试目录与虚拟环境假设工程目录叫tcp_demomkdir tcp_demo cd tcp_demo python -m venv .venv在 Windows PowerShell 激活venv.venv\Scripts\Activate.ps1在 Linux/macOS activate source .venv/bin/activate激活之后命令行前面会出现(.venv)前缀。这时python和pip都指向虚拟环境里的解释器和全局环境隔离。再创建一个requirements.txt。这个测试项目其实只依赖标准库不依赖第三方包但为了演示管理流程我通常建议加入一个开发工具依赖比如# 用于测试时捕捉包依赖关系生产环境可以不装 pytest8.0,9.0然后执行安装pip install -r requirements.txt这一步虽然瞬间完成但已经走完了标准的声明依赖、安装依赖链路。等后续项目引入真正的第三方库比如 requests、psutil同样的命令同样适用。6.2 在 Windows 下借助 iPerf 或 Python 脚本互测连通性如果你的目标不只想验证自己代码还想验证操作系统 TCP/IP 协议栈本身这里有两个思路第一个是纯 Python 脚本互测。你前面已经有 echo server/client把服务端和客户端分别跑在两个终端观察收发过程。要验证粘包与拆包可以把客户端连续发 100 条消息服务端用recv_exact按长度完整解析。第二个是借助 iPerf 做端到端的带宽与连通性测试。iPerf 是一款常用于测试最大 TCP/UDP 带宽的工具在 Windows 上可以直接下载可执行文件在 Linux 上可以用包管理器安装。用法是服务端iperf3 -s客户端iperf3 -c 127.0.0.1如果本机回环测试通过再把-c后面的 IP 换成另一台设备的局域网 IP即可验证跨机器 TCP 链路。iPerf 的好处是它自带完整的 TCP 会话管理、窗口调优和带宽统计测试协议栈连通性比你自己写压测脚本更可信。提示在 Windows 用 iPerf 时如果客户端连不上服务端先用telnet 127.0.0.1 5201验证端口是否可达。如果 telnet 也不通多半是防火墙拦截 5201 端口打开入站规则即可。iPerf 测试的输出里有两个重点参数Retr重传次数和Transfer实际传输量。如果Retr偏高说明链路有丢包或拥塞与代码本身无关那是网络环境问题。这个区分对后续调试非常重要——你以为自己代码有 bug其实是对端网络质量不行。6.3 用netstat实测 TIME_WAIT 和端口状态测试完一遍后不要急着关终端。观察一下 TCP 连接断开后的状态迁移能帮你把协议栈状态图从纸面变成直觉。在 Windows 上执行netstat -an | findstr 8080在 Linux 上执行netstat -an | grep 8080如果客户端主动关闭连接你大概率能看到TIME_WAIT状态出现在客户端侧因为主动关闭方要保证最后的 ACK 可能丢失时能重发。如果服务端侧出现大量CLOSE_WAIT说明服务端代码没正确关闭连接——很可能是忘记调用close()或异常路径没走 finally 块。这就能反过来验证代码质量一个好的 TCP 服务端程序在处理完每个连接后不应留下成堆的CLOSE_WAIT。出现这个状态就是要查连接生命周期管理了。7. 实战案例做一个带超时控制和简单重连的 TCP 客户端单纯 echo 还不够。实际项目里客户端面对的都是服务端可能掉线、网络可能抖动、响应可能超时的场景所以我来写一个更像是工程的客户端支持连接超时、收发超时、自动重连。# robust_client.py import socket import time class TcpClient: def __init__(self, host, port, timeout5.0): self.host host self.port port self.timeout timeout self.sock None def connect(self): self.sock socket.socket(socket.AF_INET, socket.SOCK_STREAM) # 设置 socket 级超时使 connect/recv/send 都受统一超时约束 self.sock.settimeout(self.timeout) self.sock.connect((self.host, self.port)) def send_and_receive(self, payload: bytes): self.sock.sendall(payload) # 服务端可能对同一连接返回多条数据这里先取一次 # 工程化时通常配合长度前缀循环读取 return self.sock.recv(4096) def close(self): if self.sock: self.sock.close() self.sock None def run_with_retry(self, payload, attempts5, delay2.0): for i in range(attempts): try: self.connect() data self.send_and_receive(payload) print(f第 {i1} 次尝试成功收到: {data!r}) return data except socket.timeout: print(f第 {i1} 次尝试超时) except ConnectionRefusedError: print(f第 {i1} 次尝试连接被拒绝) except OSError as e: print(f第 {i1} 次尝试网络错误: {e}) finally: self.close() time.sleep(delay) raise RuntimeError(多次重试仍然失败) if __name__ __main__: c TcpClient(127.0.0.1, 8080, timeout3.0) c.run_with_retry(bhealthcheck)这里的设计逻辑有三个值得讲的地方超时不是只设连接超时。Python 的标准做法是给整个 socket 设settimeout这样 connect、recv、send 都会受约束。如果你只给connect包装socket.create_connection((host, port), timeout3)那只能控制建立连接这一步建立之后 recv 如果一直等不到数据照样无限期阻塞。重试逻辑要区分异常类型。ConnectionRefusedError多数是服务端没监听重试意义不大socket.timeout可能是瞬态状况重试有价值。把两类错误混在一起无脑重试一是浪费时间二是无法定位问题。上面例子把异常分开打印就是为了拿到第一手信息。重连时必须重建 socket 对象。TCP 连接一旦断开或出错就不能复用了必须close()掉旧对象下次connect()时开个新的。这个细节不写出来的话同一套代码跑第二次必报OSError: [Errno 107] Transport endpoint is not connected。8. requirements.txt 最佳实践落地清单这一章集中说结论直接照着抄即可。但每条背后都有对应场景不是单纯的死板规定。8.1 完整的项目文件结构示例一个可复现的 TCP 小项目我建议至少包含这些文件tcp_demo/ ├── .venv/ # 虚拟环境不提交 ├── .gitignore # 屏蔽 .venv、__pycache__、.pytest_cache ├── requirements.txt # 直接依赖声明 ├── requirements.lock # pip freeze 生成的部署锁文件 ├── server_echo.py ├── server_length.py ├── server_threaded.py ├── robust_client.py └── README.md.gitignore至少写这几行.venv/ __pycache__/ *.pyc .pytest_cache/8.2 requirements.txt 标准模板# 核心依赖 requests2.31.0,3.0.0 # 测试依赖 pytest8.0,9.0原则一手动维护只写直接依赖禁止直接pip freeze requirements.txt。原则二提供范围而不是精确锁定精确锁定看下一节的 lock 文件。原则三使用-r实现多环境继承。8.3 生成部署用 lock 文件在开发环境最好是 Linux 或与生产环境一致的系统执行pip freeze requirements.lock这个文件用来精确锁定所有直接和间接依赖的版本提交到仓库中。部署环境安装用pip install -r requirements.lock注意lock 文件不能替代 requirements.txt因为它的版本锁定会让日常开发升级困难。二者搭配前者保证开发期依赖演进灵活可控后者保证部署期环境完全一致。8.4 升级依赖版本的正确姿势升级一个包前先查看它当前的关系树pip install pipdeptree pipdeptree --packages requests然后修改 requirements.txt 版本范围再用 venv 全新安装测试python -m venv .venv_test新环境下跑一遍用例通过后再把测试环境删除。养成改范围、建新环境验证、再提交的习惯能显著降低升级引入的不兼容风险。很多线上事故不是代码改炸的是依赖升级没验证就上了。9. 把网络编程项目做成工程的关键收尾写到这里再碎碎念几个我踩过的坑以及现在怎么回避它们的。首先协议设计永远先于代码实现。写 socket 之前先把消息格式定下来字段是什么类型、长度用几个字节、字节序是大端还是小端、错误码怎么定义。把这些写进文档或注释里不然三天后连自己都看不懂这个 b\x00\x05\x00 是什么意思。其次日志比 print 更重要。网络程序的时序问题极难直观看到建议直接用标准库logging分散到各模块记录连接建立、断开、重试、异常。生产环境中我还习惯给每条 TCP 连接分配一个连接 ID日志里持续带着这个 ID排错时能把多端行为串起来。再次不要把socket对象当作线程安全的。多个线程同时对一个 socket 进行 send 或 recv 会产生数据交叉和混乱。如果必须多线程收发建议用队列把数据排队由一个线程统一执行 socket 写操作。最后是关于 TCP_NODELAY 的两三句。Nagle 算法会把小包攒到一起发送在需要低延迟的交互场景比如你请求一个字节、等一个响应会造成明显延迟。可以用sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)关闭。但注意关闭后网络包数量会变多带宽消耗上升。高吞吐批量传输场景反而别关让 Nagle 帮你合并小包更划算。依赖管理这边我现在的习惯是仓库里永远放一个requirements.txt里面的版本范围只写大版本边界发布前在干净环境里跑通再生成一份 lock 文件交给 CI 用。环境切换、容器部署只要锁文件在应用行为就基本一致。在我看来Python 网络编程和requirements 管理是一枚硬币的两面前者解决的是程序如何与操作系统网络栈协作后者解决的是这个程序如何在不同机器上稳定复现。你在本机能跑、换台机器就崩的项目绝大多数不是 socket 代码的问题而是环境没有按规则还原。把消息边界、超时控制、连接生命周期和依赖清单这四件事认真对待任何规模的 TCP 项目都不会走向玄学调试。最后分享一个我自己的调试小习惯遇到网络 bug不要急着改代码先用工具telnet、iPerf、netstat确认本机到对端的 TCP 链路是健康的再回头看应用层逻辑。链路不健康改再多代码都是治标不治本。链路健康、业务仍异常那就按消息边界和超时设计逐层排查一般很快就能定位到具体那一行。