Python TCP网络编程健壮性实战与requirements.txt依赖管理最佳实践

我在排查线上服务偶发超时的时候，十次里有七次会翻车在“代码看着没毛病”的假象上——TCP 编程里真正的问题从来不在正常路径上，而在连接被对端重置、半包粘包、缓冲区溢出、依赖版本漂移这些“异常情况”里。今天这篇就围绕两个核心主题展开：一是用 Python 做 TCP/IP 网络编程时，怎么把健壮性从口号落到代码细节；二是把 requirements.txt 从“随手 pip freeze 出来的文件”升级成正经的依赖管理方案。适合已经写过 socket 入门代码、但还没经历过生产环境毒打的 Python 开发者，也适合想把项目工程化水平往上提一个档次的朋友。

这篇东西不会去复述协议教科书里的七层模型，而是直接把 TCP 的状态机、缓冲区、粘包拆包这些底层行为，映射到Python socket API 的每一个函数调用上。你会发现很多“玄学报错”，其实都是协议机制和 API 语义在特定时序下的必然结果。

1. TCP/IP 网络编程的健壮性设计思路

1.1 从协议栈的角度重新理解 socket

很多人写 socket 代码，脑子里装的是“打开一个管道，往里塞数据”这种流式模型，这本身没错，但不够。TCP 是一个面向连接的、可靠的、基于字节流的传输协议——这句话每个教程都写，但落到代码里，真正决定程序健不健壮的恰恰是“连接”、“可靠”、“字节流”这三个词背后的机制细节。

先说“连接”。TCP 的三次握手在 socket API 里分别对应客户端 connect() 和服务端 accept() 的返回。但握手成功只代表两台机器的协议栈达成了同步，不代表对端应用进程一定活着。我见过不少服务端程序只在 accept 时打印一条日志，之后对端进程崩溃了也不自知，直到下一次 write 才收到 BrokenPipeError。所以在设计健壮系统时，必须把“TCP 连接存在”和“对端应用可用”这两件事分开判断，靠的是心跳或者业务层超时，而不是 TCP 本身。

再说“字节流”。TCP 没有消息边界，这是初学者最容易踩的坑。你 send 了两次数据，对端 recv 一次可能全部收到，也可能分三次收到，这跟对端缓冲区大小、网卡中断时机、内核调度都有关系。要实现可靠的消息通信，必须在应用层自己定义消息格式——常见做法是长度前缀法，也就是每个消息前固定加 4 字节的长度信息。这块我在第 3 章详细展开。

最后说“可靠”。TCP 的可靠是“尽力投递、超时重传、最终一致”，它保证的是字节流的顺序和完整性，不保证实时性。所以 TCP 连接可能会在一段时间没有数据传输后被中间设备（比如 NAT、防火墙）静默回收，表现就是连接看起来还在，一 write 才发现已经死了。这是半开连接问题，后面会给出检测方案。理解这三点，写出来的 socket 代码才会考虑连接生命周期、消息边界和空闲检测，而不是只盯着收发函数本身。

1.2 健壮性到底指什么：异常路径的防御设计

我在评审代码时经常问一个问题：如果对端在 recv 返回了 0 字节，你的程序会怎么样？这个场景就是 TCP 四次挥手的 FIN 包到达时，recv 函数的返回值。很多人没处理这个分支——0 字节返回表示对端已经关闭了写端，这是正常关闭信号，不是错误。但如果你把它当成普通数据循环继续读，就会陷入死循环空转，CPU 打满。

健壮性的核心就是对异常路径的覆盖，具体到网络编程，主要包含这几类：

• 连接建立阶段的异常：目标不可达、超时、连接被拒绝。
• 数据传输阶段的异常：半包粘包、对端重置连接、发送缓冲区满。
• 生命周期管理的异常：半开连接、对端优雅关闭与强制关闭的区分。
• 资源层面的异常：文件描述符耗尽、内存缓冲无限增长。
• 依赖环境层面的异常：操作系统 socket 参数限制、库版本行为漂移（比如 requests 的行为在不同版本间有破坏性变更）。

一个健壮的网络程序，正常路径上的代码可能只占 30%，剩下 70% 都在处理这些事情。你在设计阶段就要把这些分支画出来，然后逐个实现对应的兜底策略。下面整个第 3 章的示例代码，就是围绕这些异常路径展开的。

2. 工具选型与核心原理解析

2.1 Python socket API 的关键细节盘点

Python 的 socket 模块是对 BSD socket API 的薄封装，所以 C 语言里那些坑它一个不落。有几个细节非常关键，直接决定代码质量。

第一个是 send() 的返回值问题。很多人以为 send() 一次会发送完所有数据，但在非阻塞模式下，它只保证尽量发送，返回值可能小于你传入的长度。阻塞模式下通常能发完，但依然不应该依赖这一点。规范的写法是循环发送，直到所有数据都发出，或者直接用 sendall()。但 sendall() 也不是银弹——它其实是在帮你做循环，如果中途连接断了，它照样抛异常。

第二个是 recv() 的缓冲区语义。recv(bufsize) 的 bufsize 只是本次调用的最大读取量，不能当作消息边界。如果对端发送的数据大于 bufsize，剩余数据仍然留在内核缓冲区里，下次 recv 接着读。如果对端发送的数据小于 bufsize，recv 只返回实际读取的长度。所以返回值需要仔细处理，如果你把 bufsize 设得过大，比如 64KB，而消息平均只有几百字节，内存和效率都不理想。生产环境一般建议 4096 到 8192 这个量级，消息体大的再适当调大。

第三个是 settimeout() 与阻塞模式的关系。socket.settimeout(value) 设置了 socket 的 I/O 超时，一旦设置，该 socket 就处于“带超时的阻塞模式”，内部会把socket切换到非阻塞然后自己用 select 实现等待。超时后调用会抛 socket.timeout 异常，这是 OSError 的子类，捕获时要注意顺序。设计上，connect 和 recv 的超时时间应该不同，connect 可以用较短的 3-5 秒，recv 则根据业务容忍度放长一些。

第四个是 shutdown() 和 close() 的区别。close() 只是把文件描述符的引用计数减一，如果还有其他地方持有引用，连接不会真正关闭；而且 close 立刻返回，不保证 FIN 包已经发出。shutdown(SHUT_WR) 会发送 FIN，表示“我不会再发送数据了”，但还能接收。优雅关闭的标准姿势是：先 shutdown(SHUT_WR)，然后继续 recv 直到收到 0 字节（对端也关闭了写端），最后才 close。这套流程保证了四次挥手完整走完，数据不会丢失。

2.2 requirements.txt 的角色定位与版本控制哲学

requirements.txt 不是一个“可有可无的文件”，它是 Python 项目可复现性的根基。这个文件的核心使命就一句话：让项目在任何环境、任何时间点都能装出相同的依赖集合。

这里有两个层次。第一层是直接依赖，就是你的代码里 import 的那些库；第二层是传递依赖，就是这些库自身依赖的库。很多新手用 pip freeze > requirements.txt，这个命令会把当前虚拟环境里所有已安装的包全部导出来，包括那些你根本没用到的传递依赖。这种方式有两个问题：一是噪音太大，文件里一堆无关包；二是没有区分直接依赖和传递依赖，一旦上游库更新，你无法判断该升级什么。

更合理的做法是用 pip-tools 维护两个文件：requirements.in 记录顶层依赖及版本约束，比如 requests>=2.25,<3.0；requirements.txt 则通过 pip-compile 生成，里面是所有依赖的精确锁定版本，包括传递依赖。升级时改 .in 文件，重新 compile 即可。这套工作流相当于把 Maven 或 npm 的依赖管理体验搬到了 Python 生态。

还有一个容易忽视的细节：requirements.txt 的版本号不要用 >= 裸奔，尽量用 == 精确锁定。因为你无法预知未来某个小版本会不会改变行为。我实际遇到过 urllib3 一个 patch 版本升级后连接池行为变化，导致线上并发请求排队时间暴增的事故。生产环境里，“不变”本身就是一种重要的稳定性。

3. 实操过程：从零实现一个健壮的 TCP 通信模块

3.1 设计一个带长度前缀和心跳的消息协议

直接上代码。设计思路是：每条消息由 4字节大端长度头 + JSON载荷 组成，长度指的是 JSON 部分字节数，不含头部。加心跳包属于控制消息，payload 是一个字典 {"type": "heartbeat"}，上层业务通过 type 字段区分。

python复制import json
import socket
import struct


def encode_message(obj: dict) -> bytes:
    payload = json.dumps(obj, ensure_ascii=False).encode("utf-8")
    header = struct.pack("!I", len(payload))
    return header + payload


def decode_message(buffer: bytearray) -> tuple[dict | None, bytearray]:
    # 从缓冲区中解析一条完整消息，返回 (消息, 剩余缓冲区)
    if len(buffer) < 4:
        return None, buffer
    (msg_len,) = struct.unpack("!I", bytes(buffer[:4]))
    if len(buffer) < 4 + msg_len:
        return None, buffer
    payload = bytes(buffer[4:4 + msg_len])
    obj = json.loads(payload.decode("utf-8"))
    del buffer[:4 + msg_len]
    return obj, buffer

注意这里用了 bytearray 作为累积缓冲区，del buffer[:4 + msg_len] 是原地切片删除，避免频繁创建新对象。这个函数的核心逻辑就是把 TCP 字节流重新切成一条条消息——处理粘包时，第一次 recv 可能拿到多条消息，循环解析直到缓冲区不够一个完整头部；处理半包时，缓冲区不足 4 字节或不足声明的长度，就继续等待更多数据。

为什么长度头选 4 字节大端？因为 4 字节无符号整数最大能表示约 4GB，足够绝大多数消息场景；大端序（!I）是网络字节序，跨平台不打架。JSON 作载荷格式是为了调试方便，性能敏感场景可以换 protobuf 或 msgpack，协议头部分不用改。

3.2 服务端实现：并发模型与优雅关闭

服务端我用 socketserver 改造还是纯 socket 手写？我选择纯手写，因为框架层会让异常处理变得隐晦，不利于展示健壮性细节。下面是一个线程池版的服务端核心代码，处理连接、分包、业务分发和优雅关闭。

python复制import socket
import threading
import queue
from concurrent.futures import ThreadPoolExecutor


class TCPServer:
    def __init__(self, host: str, port: int, workers: int = 8):
        self.host = host
        self.port = port
        self.pool = ThreadPoolExecutor(max_workers=workers)
        self._stop_event = threading.Event()
        self._clients = set()
        self._client_lock = threading.Lock()

    def start(self):
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        # 允许端口复用，解决 TIME_WAIT 状态下重启失败的问题
        self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.sock.bind((self.host, self.port))
        self.sock.listen(128)
        self.sock.settimeout(1.0)
        print(f"server listening on {self.host}:{self.port}")
        while not self._stop_event.is_set():
            try:
                conn, addr = self.sock.accept()
            except socket.timeout:
                continue
            except OSError:
                break
            conn.settimeout(30.0)
            with self._client_lock:
                self._clients.add(conn)
            self.pool.submit(self._handle_client, conn, addr)

    def _handle_client(self, conn, addr):
        buffer = bytearray()
        try:
            while True:
                chunk = conn.recv(8192)
                if not chunk:
                    break
                buffer.extend(chunk)
                while True:
                    msg, buffer = decode_message(buffer)
                    if msg is None:
                        break
                    self._dispatch(conn, msg)
        except socket.timeout:
            print(f"client {addr} recv timeout, closing")
        except ConnectionResetError:
            print(f"client {addr} reset connection")
        except Exception as e:
            print(f"client {addr} error: {e!r}")
        finally:
            with self._client_lock:
                self._clients.discard(conn)
            try:
                conn.shutdown(socket.SHUT_RDWR)
            except OSError:
                pass
            conn.close()
            print(f"client {addr} disconnected")

    def _dispatch(self, conn, msg):
        if msg.get("type") == "heartbeat":
            return
        # 业务处理，如需回写响应使用 conn.sendall(encode_message(...))
        print(f"recv msg: {msg}")

    def stop(self):
        self._stop_event.set()
        with self._client_lock:
            for conn in list(self._clients):
                try:
                    conn.shutdown(socket.SHUT_RDWR)
                    conn.close()
                except OSError:
                    pass
        self.sock.close()
        self.pool.shutdown(wait=False)

几个关键点说明一下：listen(128) 设置了 accept 队列长度，高并发场景这个值太小的直接后果是客户端 connect 成功但服务端 accept 不过来，表现为连接建立慢或者 ECONNREFUSED。SO_REUSEADDR 是服务端必备，否则服务重启时端口可能被上一个进程的 TIME_WAIT 状态占着，bind 直接失败。accept() 设 1 秒超时是为了让主循环能定期检查 _stop_event，实现可控停机——这是优雅关闭的基础。

在 _handle_client 里，把每个连接上的 socket 也设置了 30 秒超时，这是为了防止某个客户端半死不活地挂着又不主动断开，导致线程池被占满。超时杀连接虽然粗暴，但在资源隔离这个维度上是对的：一个异常客户端不应该拖垮整个服务。

stop() 方法先通过 _stop_event 让 accept 循环退出，然后遍历所有活跃连接，逐个 shutdown 并 close。注意 shutdown 可能抛 OSError，因为对端可能已经关闭了，此时忽略异常就好。这个停机能保证即使有客户端正在传输数据，服务端也能在几百毫秒内完成断开，不会出现进程退出了端口还被占着的情况。

3.3 客户端实现：重试、超时与半开连接检测

客户端这边的健壮性设计比服务端更复杂，因为它是主动方，所有失败都要做决策：是重试还是报错，重试多少次，每次间隔多久。

python复制import socket
import time
import random


class TCPClient:
    def __init__(self, host: str, port: int, connect_timeout=5.0, read_timeout=10.0):
        self.host = host
        self.port = port
        self.connect_timeout = connect_timeout
        self.read_timeout = read_timeout

    def connect(self, retries: int = 3, backoff: float = 0.5):
        last_err = None
        for attempt in range(retries):
            try:
                sock = socket.create_connection(
                    (self.host, self.port),
                    timeout=self.connect_timeout,
                )
                sock.settimeout(self.read_timeout)
                # 禁用 Nagle 算法，降低小消息的延迟
                sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
                return sock
            except socket.timeout:
                last_err = f"connect timeout (attempt {attempt + 1})"
            except ConnectionRefusedError:
                last_err = f"connection refused (attempt {attempt + 1})"
            except OSError as e:
                last_err = f"os error: {e!r} (attempt {attempt + 1})"
            # 指数退避 + 随机抖动，避免重试风暴
            sleep_time = backoff * (2 ** attempt) + random.uniform(0, 0.1)
            time.sleep(sleep_time)
        raise ConnectionError(f"connect failed after {retries} attempts: {last_err}")

    def send_message(self, sock: socket.socket, obj: dict):
        data = encode_message(obj)
        sock.sendall(data)

    def recv_message(self, sock: socket.socket) -> dict:
        buffer = bytearray()
        while True:
            msg, buffer = decode_message(buffer)
            if msg is not None:
                return msg
            chunk = sock.recv(8192)
            if not chunk:
                raise ConnectionError("connection closed by peer")
            buffer.extend(chunk)

    def ping(self, sock: socket.socket) -> bool:
        """发心跳检查连接是否真的可用"""
        try:
            self.send_message(sock, {"type": "heartbeat"})
            resp = self.recv_message(sock)
            return resp.get("type") == "heartbeat"
        except (socket.timeout, OSError, ConnectionError):
            return False

create_connection 是官方推荐的连接函数，它内部会做 DNS 解析、多地址逐个尝试（IPv4/IPv6），比自己 socket() 再 connect() 更靠谱。重试策略用了指数退避加随机抖动，这是为了防止大量客户端同时失败后，重试请求在时间上聚集成新的尖峰——没有抖动的固定间隔重试，本质上就是在给自己制造雪崩。

TCP_NODELAY 禁止了 Nagle 算法。Nagle 会把多个小包合并后再发送，这在交互式低延迟场景（比如心跳）里是致命的，可能让一个 30 字节的心跳在极端情况下延迟几十毫秒。代价是网络上的小包数量增多，对于局域网内部服务来说，这个取舍很划算。

ping() 方法用来检测半开连接：空闲很久后拿不准连接是否还活着，就发一个心跳。如果 write 都没报错，但长时间等不到响应，基本可以判定连接已经失效，该走重连流程了。这个函数可以配合业务层的健康检查定期调用。

3.4 实操演示：本地回环联调与 iperf 验证

代码写完，第一步就是用本地回环地址做联调。回环测试隔绝了物理网卡和交换机的干扰，能快速定位自己代码里的问题。启动服务端后，再用一个简单的命令行客户端发几轮消息验证。这时可以开两个终端，一个 tcpdump 或者 Wireshark 抓包，观察 TCP 握手、数据段和 FIN 包的时序，对协议细节会非常有体感。

如果你没有抓包工具，可以装 iperf 来验证网络链路的带宽和丢包基线——这在排查网络问题时价值很大。先确认服务端机器到客户端机器的带宽瓶颈是不是在应用代码之外。比如用 iperf -s 跑在服务端，iperf -c <server_ip> -t 10 跑在客户端，如果底层链路只有 10Mbps，你应用层怎么调优都白搭。这一步是排查问题的基础，至少能帮你区分“链路问题”和“代码问题”。

我在本地联调时的核对清单：

• 服务端能 accept 客户端连接，握手包正常。
• 客户端发 1000 条消息，服务端 recv 到 1000 条，不多不少，验证没有粘包拆包错乱。
• kill 客户端进程（强制关闭，而不是优雅关闭），服务端能在 30 秒超时后打印超时日志并清理连接，而不是卡死。
• 第二次启动服务端不报 address already in use。
• 连续启动、停止服务端 10 次，没有文件描述符泄漏（监控进程 fd 数量）。

这些检查项都过了，服务端的单机健壮性才算及格。

4. requirements.txt 最佳实践：从裸奔到可复现

4.1 pip freeze 的坑与 pip-tools 的正确姿势

很多项目的 requirements.txt 就是这么来的：

bash复制pip freeze > requirements.txt

然后提交到 git，完事。这个命令的问题在于，它不仅导出了你直接依赖的包，也导出了所有传递依赖、甚至虚拟环境里残留的无关包。它不区分这些包是项目的业务依赖还是某个包的附属依赖。一旦哪天有人 pip install -r requirements.txt，装出来的环境版本组合可能和开发环境不一致，因为某个直接依赖根本没被记录（如果你当时是用 pip install 装的但没主动 freeze），或者某个传递依赖的更新版本引入了不兼容的行为。

用 pip-tools 的方式，维护两个文件。

requirements.in 内容示例：

code复制requests==2.31.0
Flask==3.0.0
redis==5.0.0
pika==1.3.2

执行：

bash复制pip install pip-tools
pip-compile requirements.in

生成的 requirements.txt 里是所有依赖的精确版本，并且带上了 via 注释标明每个包是被哪个包引入的。后续环境安装就用：

bash复制pip install -r requirements.txt

需要升级依赖时，改 requirements.in 里的版本号，再重新 pip-compile。整个过程干净利落，requirements.txt 变成了一种可以审查、可以追踪变更的产物，而不是一个垃圾堆。

还有一个加分的操作是把编译产物纳入版本控制，并且在 CI/CD 流水线里用 pip install -r requirements.txt 构建。这样生产环境的依赖集合一定和测试环境一致，从源头消灭“我本地是好的啊”这类经典甩锅语句。

4.2 分环境管理：开发、测试、生产的依赖分层

很多团队只维护一个 requirements.txt，把所有依赖打成一个包。这在早期没问题，但项目复杂之后，开发环境有 dev tools（pytest、black、mypy），生产环境不需要这些东西，硬塞进去不仅拖慢安装速度，还可能引入安全隐患。

推荐的拆分方式：

text复制requirements/
├── base.txt          # 生产/所有环境共有的依赖
├── dev.txt           # 开发环境额外依赖，如 pytest、black、ruff
└── prod.txt          # 生产环境额外依赖，如 gunicorn、gevent

base.txt 通过 -r base.txt 被另外两个文件引用。比如 dev.txt 第一行写 -r base.txt，然后追加开发工具；prod.txt 也一样。安装时 pip install -r requirements/dev.txt 就能装齐开发环境的所有依赖。

这个分层的好处有几点：生产环境的安装时间更短、攻击面更小（不会安装不必要的包）；开发环境可以放心添加 linter 和测试框架，不污染生产基线；依赖变更的 review 范围更小，diff 一清二楚。我见过不少团队把 pytest 装进了生产容器，虽然是小事，但对镜像体积和安全性都是不必要的负担。

4.3 版本锁定粒度与安全更新节奏

版本锁定用什么符号，是个值得说道的事情。>= 太宽松，== 太死板，~= 又没那么多人熟悉。我推荐在 .in 文件里用兼容范围，在编译生成的 .txt 文件里用精确版本。前者负责定义你的意图，后者负责锁定现实。

举个例子，requests==2.31.0 与 requests>=2.25,<3.0 的区别：前者是“我只要这个版本”，后者是“我要 2.x 系列，但不能跳到 3.0 的破坏性版本”。如果用前者，每次升级都要手动改文件跑 pip-compile，容易被忽略；如果用后者，每隔一段时间 pip-compile 就会尝试拉取范围内最新版本，你能及时获得安全补丁，同时有 <3.0 的上限做保护。

但安全更新不能完全靠被动等。这里有两个实践建议：一是用 pip-audit 之类工具扫描依赖漏洞，跑在 CI 里，检测到高危漏洞就阻断合并；二是给依赖升级单独建立 PR，不要和功能开发混在一起，否则出了问题不好追溯。安全更新节奏上，我的习惯是每周固定花一点时间做一次 pip-compile --upgrade，然后跑全量测试，看是否有回归。这个节奏既不会太频繁让人崩溃，也不会太松散让漏洞堆积。

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

5.1 高频异常速查表

5.2 实战排障过程：一个坚持了七天的并发问题

我印象很深的一个案例是：服务上线后头几天一切正常，到第七天突然开始大面积请求超时，cpu 和内存都正常，但 qps 掉了一半。

先查了连接数，ss -s 显示 TIME_WAIT 连接数量爆炸。问题出在我们的客户端没有做连接复用，每次业务请求都重新 connect 一次，造成大量 TIME_WAIT 状态的套接字躺在内核里。而 net.ipv4.ip_local_port_range 默认的可用端口范围是有限的，短时间内大量短连接把可用本地端口耗尽，新连接无法分配端口，表现就是 connect 超时或直接被拒。

当时没有立刻上连接池，第一刀先优化了代码里两处不需要每次新建连接的路径，问题立刻缓解。后续把连接池加进去，彻底根治。这个案例的教训是：TCP 连接本身不是免费的，频繁建立短连接对内核资源消耗远超预期。在网络编程里，连接池不是“性能优化项”，而是“必做项”，尤其是在高 QPS 场景下。

5.3 排查工具与方法论

排查网络问题的工具就那么几个，但用得好能省一半时间。

ss -lntp 是看监听端口和连接状态的利器，比 netstat 输出更清晰、更快。ss -s 可以看整体连接状态统计，上手先跑一下，能快速判断是不是端口耗尽、TIME_WAIT 堆积这类宏观问题。

抓包工具我推荐 tcpdump 命令行版本，直接在服务器上用，不用开 GUI。抓 HTTP 或自定义 TCP 协议的示例如下：

bash复制tcpdump -i eth0 host 10.0.0.5 and port 8080 -w capture.pcap

抓完拿回本地，用 Wireshark 打开分析握手包、重传包、RST 包。看到一个大量重传的记录，就能判断是链路丢包；看到一个 RST 跟在 PSH 后面，就能锁定是应用层主动断开。

最后是日志。网络编程的日志一定要带上连接标识、操作类型和耗时。我习惯在每次 recv/send 的边界打点：连接建立、连接关闭、超时事件、异常堆栈，全部带上对端地址和时间戳。没有日志的排查等同于盲人摸象，上线前一定要把日志埋好。

5.4 预防性设计清单

经验攒多了，我发现健壮性不是靠事后救火，而是靠设计阶段的一纸清单。整理一份我每次写网络服务都会过一遍的检查项，供参考：

• 所有 socket 是否设置了超时，包括 accept、connect、recv、send。
• 是否处理了 recv 返回 0 字节的分支。
• 消息协议是否有明确边界（长度前缀或分隔符），是否在文档里写清楚了。
• 服务端 accept 循环是否有退出机制，是否能优雅停机。
• 客户端连接是否做了超时和重试，重试是否带退避和抖动。
• 连接是否池化，池的大小是否根据并发量做过估算。
• 所有打开的资源（socket、文件、锁）是否都有确定的关闭路径，是否都放在 finally 或 with 里。
• 是否监控了 fd 数量、TIME_WAIT 数量、连接数、消息积压量。
• 依赖文件是否锁定版本，升级是否有独立流程和回归测试。

这九个问题，在代码 review 时逐条打勾，能拦截大多数线上事故的苗头。就拿超时这一项来说，多少线上故障归根结底是某个 socket 操作没有设超时，导致线程被永久卡住。一个 settimeout(5) 就能解决的事，缺了它就是事故。

我个人在实际操作中的体会是，网络编程的健壮性从来没有一蹴而就的解法，它是在一次次事故复盘里修正出来的。TCP/IP 协议栈复杂，但给你提供了足够多的信号——超时、重置、半关闭、零返回，每一个信号都在告诉你发生了什么，你需要做的就是尊重这些信号，并给每一个分支设计合理的响应。而 requirements.txt 这种看似不起眼的工程文件，恰恰是项目管理里“确定性”的具象化。把这两件事做到位，Python 服务的稳定性就能上一个实在的台阶。