前段时间帮一个朋友排查他那边的一个"文件传输偶尔损坏"问题，程序本身逻辑看着没什么毛病，就是用 TCP通讯 发文件，发送端把文件读出来 write 进 socket，接收端 recv 到就写盘，但传大文件总是时不时少几十 KB。聊到最后发现是字节流边界处理出了问题，他直接把"发送完整文件"当成了"一次 write 就能表示一个完整文件"，数据一多就乱了。这个场景其实特别典型，这篇文章就基于这个实际问题，把 TCP 传输文件的完整设计、代码实现、常见坑和生产化改造一次说清楚，适合正在写传输模块、又不想被粘包半包折磨的开发者。
1. 目标拆解：在 TCP 字节流中定义"文件"
很多人在初学阶段会有个直觉：TCP 是可靠传输，我只要把文件内容发过去，对面一片一片收就能拼回去。这个直觉漏掉了一个关键性质：TCP 是流式协议，不是消息协议。
1.1 为什么文件会"缺斤少两"
应用层调用 write 时，发送方的协议栈并不保证一次 write 的数据会以同样的边界到达接收方。内核可能会把两次 write 的数据合并成一个 TCP 段发送（粘包），也可能把一次 write 的数据拆成多个 TCP 段发送（半包），接收方每次 recv 返回的字节数完全没有固定规律。
比如发送方想表达"这是一个文件"这件事，如果单纯把文件内容一股脑往 socket 里塞，接收方收到的字节里没有分界信息，根本无法判断哪个字节属于文件名、哪个字节属于文件内容、哪个字节属于下一个文件。对于小于一个 MSS 的小文件，也许一次发送刚好碰到一次接收，看起来"正常"了，但只是运气好而已。
所以核心结论很简单：你必须在应用层自己定义帧协议，明确告诉接收方"帧从哪里开始、头有多长、元数据多长、载荷多长"。
1.2 需要携带哪些元数据
从使用需求出发，传输文件的接收端最少要知道三件事：
- 文件名叫什么，用于本地落盘命名。
- 文件多大，用于预分配空间、显示进度、判断是否收完。
- 数据怎么校验，用于确认传输过程没有损坏。
在此基础上，协议设计最好能解决另一个问题：接收方读到的第一个字节如何定位帧头。如果接收方刚好从一个字节流的中间开始读取（例如程序崩溃重连、多路复用同一条长连接），没有帧头标记就无法重新对齐。所以帧结构里加一个魔数（magic number）是最稳妥的做法。
1.3 自己定义协议栈的顺序
我通常会直接把结构定义成"帧头 + 元数据 + 载荷"三段式，帧头固定长度、固定布局：
- 帧头（固定部分）：2 字节魔数，1 字节版本号，2 字节文件名长度，8 字节文件大小。
- 文件名：UTF-8 编码，长度由帧头里的文件名长度描述。
- 载荷：原始文件二进制内容。
帧头固定长度是 13 字节，这个设计让接收端的解析跳过了"扫描找头"的复杂流程：先 recv 满 13 字节，解析出文件名长度和文件大小，再按需接收后续部分。如果程序要从上一条连接残留的脏数据中恢复，只需要暴力搜索魔数重新对齐。
帧结构示意（固定头）： [0 ~ 1] Magic，可以填 b'\xAB\xCD' [2] Version，目前填 1 [3 ~ 4] FileNameLen，大端 uint16 [5 ~ 12] FileSize，大端 uint64 [13 ~ ] FileNameBytes，长度 = FileNameLen [之后] FileContent，总长度 = FileSize为什么用大端而不是小端？跨语言、跨平台实现时，大端序避免了"对方平台字节序不同导致解析出错"的问题。绝大多数网络协议都统一使用网络字节序，也就是大端，这个习惯值得保留。
2. 发送端实现：先发帧头，再分块发正文
发送端的代码逻辑看起来就三步：打开文件、发帧头、发文件内容。但如果只是按直觉写，很容易踩到两个问题：那就是"文件名长度算错"和"一次性读入整个文件到内存"。这里给出一个可以直接落地的 Python 实现。
import json import os import socket import struct def send_file(sock: socket.socket, file_path: str, chunk_size: int = 64 * 1024) -> None: file_name = os.path.basename(file_path) file_size = os.path.getsize(file_path) # 1. 构造帧头并发送 file_name_bytes = file_name.encode('utf-8') header = struct.pack('>HBHQ', 0xABCD, 1, len(file_name_bytes), file_size) sock.sendall(header) sock.sendall(file_name_bytes) # 2. 分块发送文件内容 sent_bytes = 0 with open(file_path, 'rb') as f: while sent_bytes < file_size: data = f.read(chunk_size) if not data: break sock.sendall(data) sent_bytes += len(data) print(f'文件发送完成，共 {sent_bytes} 字节。')这里有两个细节值得展开讲。
第一个细节是 sendall 的语义。sendall 表示"如果数据没发完就继续发，直到全部交给内核发送缓冲区"，它会帮你处理底层部分发送的情况。如果使用 send，则必须检查返回值，循环处理剩余部分。实际开发中我几乎只用 sendall，不是因为"它一定不会丢"，而是因为如果传输中断、对端关闭，sendall 会抛出异常，程序能立刻感知到错误而不是假装成功。
第二个细节是分块读取。把整个文件一次读入内存再发送，看起来代码更简洁，但对大文件而言十分危险。一个 2 GB 的文件对应 2 GB 的 Python bytes，内存立刻爆掉，而且内核的 socket 缓冲区本身也只有一两百 KB，一次 send 大量的数据只会让协议栈内部再切片，不会带来性能提升。这里 chunk_size 取 64 KB，是一个与网络吞吐、磁盘 IO 都比较平衡的经验值。如果跑在局域网且延迟非常低，可以尝试 256 KB 或 512 KB 做基准测试对比。
2.1 Nagle 算法对传输的影响
发送端还有一个容易被忽略的底层机制：TCP 默认开启了 Nagle 算法，它会把多个小包合并后再发出，减少网络中微小数据包的数量。听起来这是好事，但和延迟敏感场景结合时会带来额外延迟。
对于一个文件传输协议，帧头只有 13 字节，在 Nagle 开启的状态下，接收方等待 13 字节这种小请求如果携带了 ACK，发送方可能要等到确认后才会把剩余数据发出，引发"确认延迟"问题。最直接的解决方案是在 TCP 握手完成后设置 TCP_NODELAY 选项，禁用 Nagle 算法。
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)这个选项对于文件传输这类大块数据场景通常是有益的。真正会受影响的是大量极短交互请求的场景，场景不同，决策方向也不同。文件传输这里我基本无脑开启。
2.2 大文件是否需要压缩
经常会有人问"传输之前要不要压缩一下"。这个问题的答案是：分场景。
- 如果文件本身是已压缩格式，例如 JPEG、MP4、ZIP，再次压缩几乎无损空间收益，反而白白消耗 CPU。
- 如果文件是日志文本、JSON、CSV 这类高可压缩性内容，压缩收益非常大，有时候能压到原文的十分之一以下。
- 如果传输链路的带宽是瓶颈，压缩是性价比极高的优化；如果 CPU 已经满负荷，压缩反而拖慢整体吞吐。
一个比较合理的做法是发送端检测扩展名或 MIME 类型，仅对特定类型尝试压缩帧。注意压缩后文件大小不再是原始大小，所以帧头里的 FileSize 要填压缩后的大小，校验也要基于压缩流。
3. 接收端实现：固定头解析与三阶段接收
接收端的代码比发送端稍微复杂一点，因为它要面对一个天然的不确定因素：recv 到底能收多少字节。很多人写接收端时习惯性认为"我 recv 一次就能拿一个完整文件内容"，这个想法必须立刻抛弃。正确的姿势是定义一个 recv_exactly 函数，循环 recv，直到收到指定数量的字节才返回。
3.1 接收端主体代码
import json import os import socket import struct def recv_exactly(sock: socket.socket, n: int, timeout: float = 30.0) -> bytes: chunks = [] received = 0 sock.settimeout(timeout) while received < n: chunk = sock.recv(min(n - received, 64 * 1024)) if not chunk: raise ConnectionError('连接已关闭，接收数据不完整。') chunks.append(chunk) received += len(chunk) return b''.join(chunks) def recv_file(sock: socket.socket, save_dir: str) -> str: # 1. 先接收固定头 header = recv_exactly(sock, 13) magic = struct.unpack('>H', header[0:2])[0] version = header[2] name_len = struct.unpack('>H', header[3:5])[0] file_size = struct.unpack('>Q', header[5:13])[0] if magic != 0xABCD: raise ValueError(f'帧头校验失败，魔数不匹配：{hex(magic)}') if version != 1: raise ValueError(f'不支持的协议版本：{version}') # 2. 接收文件名 file_name_bytes = recv_exactly(sock, name_len) file_name = file_name_bytes.decode('utf-8') save_path = os.path.join(save_dir, file_name) os.makedirs(save_dir, exist_ok=True) # 3. 接收文件内容并写入磁盘 received_size = 0 with open(save_path, 'wb') as f: while received_size < file_size: data = recv_exactly(sock, min(file_size - received_size, 64 * 1024)) f.write(data) received_size += len(data) print(f'文件接收完成：{save_path}，{received_size} 字节。') return save_path这段代码的逻辑很清楚：分三次接收不同类型的数据。第一次收固定头，第二次收文件名，第三次循环收文件内容。头解析完成后，循环调 recv_exactly 而不是直接调 recv，因为文件内容可能很大，TCP 分段到达，每次 recv 拿到的字节数是不确定的。只要循环到自己统计的 received_size 等于文件大小，就认为这次传输结束了。
3.2 为什么文件大小用流程控制而不是连接关闭
有朋友问过这样一个问题：既然 TCP 连接在传输完成后会关闭，为什么不在对端关闭连接时判断"连接关闭就说明文件传完了"？
这个思路在"一条连接只传一个文件"的场景下能工作，但它有一个明显缺陷：一旦你想复用连接，连续传多个文件，你就无法区分哪个文件结束了。更糟糕的是，如果发送端因为异常直接断开，接收端也会认为"传输完成"，最后得到半个有效文件。
判断文件是否完整，应该完全依赖帧头里的 FileSize 字段，而不是连接状态。接收完 FileSize 字节，这个文件就算收完了，之后连接爱怎么用怎么用，可以继续传下一个文件。这个设计能让你把接收逻辑从连接生命周期中解耦出来，后续扩展多文件传输、断点续传都更容易。
3.3 路径穿越与文件名安全
接收端拿到文件名后直接拼接路径，这里有个安全漏洞：如果文件名里包含../这样的路径成分，接收端可能把文件写到目标目录之外，造成路径穿越。这个问题在公网传输场景尤其严重。
一个防御性写法是只取 os.path.basename 作为保存文件名，或者对文件名做白名单清洗，把路径分隔符、控制字符全部替换掉。
import re safe_name = re.sub(r'[\\/:*?"<>|\x00-\x1f]', '_', file_name) safe_name = os.path.basename(safe_name) or 'unnamed_file' save_path = os.path.join(save_dir, safe_name)虽然"自己的客户端不会传恶意文件名"听起来像是个合理假设，但接入公网或对接第三方设备时，这个假设就不成立了。做防御性替换成本极低，收益极高，强烈建议保留。
4. 可靠传输的最后一公里：校验与完整性验证
前面几步完成后，文件能成功从 A 传到 B，但"能传"和"传得对"是两码事。TCP 本身虽然提供可靠传输，但数据经过中间设备、操作系统缓冲区、网卡驱动等环节，理论上仍有极低概率出现位反转，而且同样存在程序 bug、连接异常中断导致内容缺少一部分的情况。接收端必须能自己验证文件的完整性，这个问题通常借助摘要算法解决。
4.1 全文件哈希校验与边传边校验
最常见的行为是传输前计算一个哈希值，例如 MD5 或 SHA256，放进帧头元数据里，接收端收完全部字节后，计算本地文件的哈希并和帧头里携带的哈希比对。如果一方算出的值不一致，立即重新传输或报错。
这里有一个细节：对大文件来说，先算哈希再发送，意味着发送端要先把整个文件读一遍，网络传输还要再读一遍，磁盘 IO 翻倍。接收端同样要算哈希，总的磁盘读取量至少是文件体积的三倍。这在文件体积在几十 GB 甚至上百 GB 时是个不可忽视的成本。
另一种做法是边传边算。发送端在分块读取的时候同时更新哈希上下文，接收端在写盘的时候同步更新自己的哈希上下文，最后只需比较两个摘要值。这样文件数据只被读一次、写一次，哈希计算顺带完成，不需要额外做两次全文件 IO。
import hashlib def file_hash_chunks(file_path: str, chunk_size: int = 64 * 1024): h = hashlib.sha256() with open(file_path, 'rb') as f: while True: data = f.read(chunk_size) if not data: break h.update(data) return h.hexdigest()4.2 校验字段放入帧头还是独立帧
我在帧头里预留一个 32 字节的哈希字段之后，遇到一个新问题：有些场景不需要校验，例如进度显示型工具；有些场景想要 SHA256，有些场景只想用 CRC32 快速检查。固定 32 字节帧头方案对一定长度的哈希兼容，但不够灵活。
更优雅的方案是定义独立元数据帧，把哈希和文件信息打包成结构化数据。具体做法是：先发一个固定头，里面只放元数据帧的总长度，然后发送一个 JSON 或二进制编码的元数据帧，里面包含文件名、大小、哈希算法、哈希值、时间戳等字段。接收端把元数据帧解析完之后，再开始接收正文。
这个设计在工程上非常实用。后续改动协议时你不需要重新定义固定头，只要在元数据帧里加字段即可。新老版本之间的兼容性也更好控制。
import json def build_metadata(file_name: str, file_size: int, digest: str, algorithm: str = 'sha256'): meta = { 'fileName': file_name, 'fileSize': file_size, 'algorithm': algorithm, 'digest': digest, 'version': 1, } payload = json.dumps(meta, ensure_ascii=False).encode('utf-8') return struct.pack('>B', 0xCC) + struct.pack('>I', len(payload)) + payload接收端先接收 5 字节（类型 + 长度），解析出元数据帧长度，再 recv_exactly 对应字节，JSON 解析后校验字段是否存在。这种变长设计在跨语言对接时依然很简单，只要双方对字段格式有共识。
4.3 校验失败的应对策略
完整性校验失败时，第一反应不是"重传整个文件"，而是先判断失败发生在哪一层。最直接的手段是分块 CRC 校验或分块哈希，让每一块数据都有独立的校验值。这样失败后你只需要重传出错的那一块，而不是把整个文件重新来一遍。
把分块校验做成一个通用抽象：报文里携带一块哈希表，每块数据 1 MB，每块对应一个 4 字节 CRC。接收方边收边校验，一旦某块校验失败，记录下来，等全部收完统一请求缺失/错误块重传。这个机制只需要在帧结构里增加工作量，代码复杂度提高一个级别，但在生产环境传输超大文件时非常有效。
5. 实际部署中必须处理的四个坑
协议和代码都写好了，真正跑起来时还会遇到一些"文档不会告诉你"的问题。以我实际调试文件传输模块的经历来说，下面这四类问题出现的频率最高。
5.1 内核缓冲区与外层 send/bind 的关系
很多人以为 sendall 发完，数据就"推"到对端了。其实 sendall 只是把数据交给本机内核的发送缓冲区，至于内核什么时候把数据封装成 TCP 段发出去，以及对端内核什么时候确认，都发生在后续时间里。如果进程随后立刻调用 close，理论上 TCP 栈会尝试把发送缓冲区的残余数据发完，但这个过程可能被 RST 中断，尤其是在对端已经关闭连接或者发送缓冲设置了超时的情况下。
更可靠的模式是：发送完所有数据后，调用 shutdown(socket.SHUT_WR) 表示"我不会再向这个连接写入数据了"，然后继续从 socket 读取对端的返回结果，直到对端关闭连接。这样能够确保所有数据到达对端后再断开。
sock.shutdown(socket.SHUT_WR) # 继续等待对端响应5.2 半打开连接与超时设置
接收方阻塞在 recv 时，如果网络设备断了一条链路而未触发 TCP 超时重传，接收方可能长时间挂着，让人误以为程序在正常工作。这个坑在弱网环境下特别明显。
体验较好的方案是：
- 设置 socket 的接收/发送超时，例如 30 秒无数据就报超时。
- 业务层加一个心跳机制，例如每 30 秒发送 4 字节心跳帧。
- 如果一个周期内既没有数据也没有心跳，按连接失效处理。
在 recv_exactly 里，每轮循环都要重设超时，而不是只在连接建立时设置一次。如果只顾设置一次，中途网络静默，判断时机可能不准。
5.3 分块传输中的进度显示与吞吐率评估
传文件的时候，进度条是一个高频需求。计算进度的坐标系统要统一，进度 = 已接收字节数 / 文件总字节数，不要用时间估算法。文件总字节数就来自帧头里的 FileSize，而 FileSize 本身就是权威数据，不要额外使用一个"文件长度缓存"。
如果想评估历次传输的吞吐率，需要关心的核心数据是：
- 连接建立耗时
- 首字节到达耗时
- 平均速度与瞬时速度
- 传输完成后的总耗时
一次完整传输的总耗时是从"开始连接"到"接收方校验通过"的整体耗时，不是数据全部发完的耗时，否则会漏掉校验阶段的读取开销。
5.4 大文件落盘与内存碎片的平衡
接收端如果使用 b''.join(chunks) 来拼接数据，内存里会保留每一次 recv 返回的分片。在局域网传输文件动辄几 GB 的场景下，这些分片会占掉令人惊讶的内存，而且 join 时会创建又一份完整的大对象，短暂的内存峰值直接翻倍。
推荐的模式是"边收边写"：收到一块，写一块。虽然磁盘 IO 次数变多，但配合 Python 的 open(..., 'wb') 默认缓冲，实际丢失的性能非常小。如果要求更低的磁盘写入次数，可以手动维护一个 1 MB 左右的写缓冲，积攒到一定量再一次性 flush，既能控制内存峰值，也不会让磁盘扛不住。
buf = bytearray() with open(save_path, 'wb') as f: while received_size < file_size: data = recv_exactly(sock, min(file_size - received_size, 64 * 1024)) f.write(data) received_size += len(data) f.flush()末尾的 flush 不是每次都必须调用，可以用 os.fsync 按需控制落盘频率。对可靠性要求极高的场景，传输完成后再 fsync 一次，确保数据真正写入磁盘而不是停留在操作系统的页缓存中。
6. 更进一步：多文件批量传输与断点续传改造
单文件传输逻辑稳定后，接下来通常有两个自然延伸：批量传多个文件和断点续传。
6.1 同一条连接连续传输多个文件
把之前的协议直接复用即可：传完一个文件不用断开连接，可以直接发送下一个文件的帧头。接收端解析完当前文件后，回到等待帧头的状态。只需要注意每个文件独立的文件名长度和大小都正确设置，加一个循环包在外面就完成了。
def send_files(sock: socket.socket, file_paths: list[str], chunk_size: int = 64 * 1024): for path in file_paths: send_file(sock, path, chunk_size) sock.shutdown(socket.SHUT_WR)这个方案的优点是不用反复建立 TCP 连接，省掉了三次握手和四次挥手的时间。批量传 100 个小文件时，效果会非常明显。
6.2 断点续传如何设计
断点续传的核心是"接收方告诉发送方我已经有的范围，发送方只发缺失的部分"。经典实现是接收方保存一个 .part 文件和一个 .meta 文件，.meta 里记录了已接收的字节数或块号。重新连接时，接收方读取 .meta，把续传点告诉发送方。
具体到协议，可以让发送方在帧头里额外增加一个偏移字段，比如 startOffset，表示这次传输从文件哪个位置开始。接收方在本地文件写入时，直接把文件 seek 到对应偏移位置，继续写后续字节。
def recv_file_resume(sock, save_path, start_offset, file_size): with open(save_path, 'ab') as f: f.seek(start_offset) received_size = start_offset while received_size < file_size: data = recv_exactly(sock, min(file_size - received_size, 64 * 1024)) f.write(data) received_size += len(data)一个容易忽略的问题：断点续传后，文件内容的哈希校验就不好做了。之前整个文件的哈希不再适用于"已有部分 + 新增部分"的组合，所以分段哈希的方案在这里优势更大。每个块独立计算哈希，续传时只校验缺失的块，逻辑清晰很多。
6.3 服务端并发接收的设计思路
在实际部署中，如果接收端是用单线程 accept，同一时刻只能处理一个连接，另一个客户端就会一直阻塞等待。最简单的并发改造是每个连接分配一个线程或协程来处理，线程池的大小需要根据预期并发连接数调整。更高效的做法是用 IO 多路复用（select / epoll / kqueue）搭配事件驱动模型，但实现复杂度会高一个量级。
对中小规模文件传输服务来说，一条连接一个线程池已经非常够用。只要注意每个线程里处理完一个文件后不要让你的循环阻塞在 recv 上，务必给 socket 设置好超时。
import threading def handle_connection(conn: socket.socket, save_dir: str): try: while True: recv_file(conn, save_dir) except (ConnectionError, socket.timeout): pass finally: conn.close() def start_server(host: str, port: int, save_dir: str): server = socket.socket(socket.AF_INET, socket.SOCK_STREAM) server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) server.bind((host, port)) server.listen(128) while True: conn, _ = server.accept() threading.Thread(target=handle_connection, args=(conn, save_dir), daemon=True).start()线程数量不要无脑拉高，否则线程切换开销会反噬吞吐量。我见过最离谱的配置是四核机器上起了 500 个线程处理转发任务，CPU 全部耗在线程切换上，而不是业务逻辑上。建议先用 2~4 倍核心数做压测，再根据 CPU 使用率和队列积压情况微调。
7. 实测数据与压测建议
写完代码只是第一步，跑出可信的实测数据才算数。我的建议是把测试分成三类：正确性测试、稳定性测试、性能测试。
7.1 正确性测试
基本路径测试是传输一个 0 字节文件、一个 1 字节文件、一个恰好等于块大小的文件、一个比块大小多 1 字节的文件、一个超过 2 GB 的文件。每一轮都要校验文件内容。如果在这五个用例中有一个通过不了，说明协议设计里存在某类边界 bug，例如零长度文件没有发送载荷、文件末尾正好落在块边界时循环提前退出等。
7.2 稳定性测试
稳定性测试的目标是"长时间跑不出错"。模拟弱网环境时，可以在中间加一层丢包率为 1% 的网卡队列或代理，线上连接因偶发丢包而重传时，你的协议要能扛住重传与乱序，不能因为事件顺序微调就异常退出。另外要模拟接收方关机、发送方重启、网络断开重连等异常中断因素，确认程序出现异常时能快速失败并清理临时文件。
7.3 性能测试的读数要点
在测试性能时，有几组数据是必须采样的：
- 文件大小跨越 1 MB、100 MB、1 GB、10 GB。
- chunk_size 分别测试 4 KB、16 KB、64 KB、256 KB、1 MB。
- 单连接传输和多连接并发传输分别记录吞吐率。
用局域网做基准测试时，如果网络本身能跑到 900 Mbps 以上，而你的程序只能跑 200 Mbps，优先排查两个方向：一是块大小是否太小，二是发送/接收逻辑里是否有个别 recv 只取少量字节导致系统调用次数爆炸。系统调用次数越少，吞吐率越高，这是一个非常直观的经验。
以 64 KB 块大小在千兆局域网上传 10 GB 文件为例，理论上数据面可以达到 900 Mbps 以上，实际工程实现通常落在 600 ~ 850 Mbps。如果远低于这个范围，多半是应用层代码有额外复制或者中间层做了不合理的逐字节处理。
用单线程测吞吐率时，还可以尝试增大 socket 缓冲区。Linux 上默认缓冲区大概在 16 KB 到 128 KB 之间，可以通过 setsockopt 把收发缓冲区调到 1 MB 或更高。
sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 1024 * 1024) sock.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 1024 * 1024)注意这个设置修改的是系统缓冲区上限值，实际生效还受内核参数 net.core.rmem_max / wmem_max 限制，调参前最好先确认一下上限。
8. 一个完整的可运行示例
为了让你能直接跑起来看效果，我放一个最小化的完整例子。它包含发送端和接收端两个函数，不依赖任何第三方库，只使用标准库就可以运行。协议采用固定头，额外附带一个 SHA256 校验字段，在传输完成后自动校验。
import hashlib import os import socket import struct MAGIC = 0xABCD VERSION = 1 CHUNK_SIZE = 64 * 1024 TIMEOUT = 30.0 HEADER_LEN = 5 + 2 + 2 + 8 + 32 # struct format: >B H H Q 32s # 帧类型: 1字节, magic: 2字节, name_len: 2字节, size: 8字节, digest: 32字节 def build_header(file_name, file_size, digest): name_bytes = file_name.encode('utf-8') header = struct.pack('>BHHQ32s', 0x01, MAGIC, len(name_bytes), file_size, digest) return header, name_bytes def parse_header(raw_header): frame_type, magic, name_len, file_size, digest = struct.unpack('>BHHQ32s', raw_header) if magic != MAGIC: raise ValueError('bad magic') return frame_type, name_len, file_size, digest def recv_exactly(sock, size): buf = bytearray() while len(buf) < size: chunk = sock.recv(size - len(buf)) if not chunk: raise ConnectionError('connection closed') buf.extend(chunk) return bytes(buf) def send_file(sock, file_path): file_name = os.path.basename(file_path) file_size = os.path.getsize(file_path) h = hashlib.sha256() with open(file_path, 'rb') as f: while True: data = f.read(CHUNK_SIZE) if not data: break h.update(data) digest = h.digest() header, name_bytes = build_header(file_name, file_size, digest) sock.sendall(header) sock.sendall(name_bytes) with open(file_path, 'rb') as f: while True: data = f.read(CHUNK_SIZE) if not data: break sock.sendall(data) def recv_file(sock, save_dir): raw_header = recv_exactly(sock, HEADER_LEN) frame_type, name_len, file_size, digest = parse_header(raw_header) file_name = recv_exactly(sock, name_len).decode('utf-8') file_name = os.path.basename(file_name) or 'unnamed' save_path = os.path.join(save_dir, file_name) h = hashlib.sha256() received_size = 0 with open(save_path, 'wb') as f: while received_size < file_size: data = recv_exactly(sock, min(CHUNK_SIZE, file_size - received_size)) f.write(data) h.update(data) received_size += len(data) if h.digest() != digest: raise ValueError('checksum mismatch, file corrupted') return save_path这个例子里的 send_file 在发送前先完整读一遍文件算哈希，接收端边收边算哈希，最终比较两个摘要。在文件体量小、追求逻辑简单的场景下，这种"先算后发"没有什么问题，直接用即可。
测试时，先在 A 终端启动接收线程，在 B 终端运行发送函数，观察控制台输出即可。如果文件很大，你可以在进度条代码里加上已接收字节数除以文件大小的百分比，效果更直观。
写到这里，TCP 传文件这个事基本完整了。作为实际使用 TCP 通讯 传输文件的经验总结，我再强调一次最值得记住的点：不要依赖 TCP 天然帮你保持消息边界，任何文件传输都必须自己设计帧协议、自己处理分块接收、自己校验完整性。这三件事做好，你在生产环境里传文件翻车的概率会降到极低。
