简介这是一份基于Python实现的可靠数据传输协议设计与源码资料包面向计算机网络课程设计、实验或相关项目实践者。内容以UDP为底层逐步实现停等协议、GBN协议并最终改进为SR协议同时支持单向、双向可靠数据传输并模拟数据包的丢失以验证协议有效性还提供了C/S结构的文件传输应用覆盖协议设计、验证与应用的完整链路。资源共14个文件以8个Python源码文件、1份Word设计报告和3个txt数据文件为主另有md与license说明文件整体压缩包仅493KB。Python源码按协议模块划分便于对照学习设计报告可用于撰写实验/课程设计文档时的思路参考txt数据文件用于测试传输效果与丢包场景模拟。目前已有1014人学习下载适合正在完成可靠传输协议相关课程作业或希望深入理解GBN、SR协议机制的读者。通过实际运行和阅读源码可掌握停等协议、滑动窗口、超时重传、选择性重传等核心要点获得一套可直接扩展的UDP可靠传输基础实现。1. 可靠数据传输协议在解决什么问题丢包重传这套戏Python 要怎么唱把一份文件从一台机器发给另一台机器网络抖了一下丢了一个包文件还能不能原样到达能只要底下有可靠数据传输协议在兜底。所谓「基于Python实现的可靠数据传输协议」就是一个在 UDP 这种只管发、不管到的传输层协议上自己动手实现确认、重传、排序的源码包。它通常是计算机网络课里最硬核的练习你拿到的 zip 拆开基本是三个角色——发送端、接收端、模拟丢包的信道。下面按这个方向走讲清楚帧怎么定、窗口怎么滑、参数怎么调照着重现实测新手能跑通熟手能避坑。如果你正好在找这类课设源码这篇文章也顺便帮你判断它的实现边界在哪里、值不值得拿来做二次开发。2. 从停等协议到滑动窗口可靠传输的三个核心机制2.1 ACK、序号与重传可靠传输的三板斧在 Python 里实现可靠传输绕不开三个机制ACK、序号、超时重传。一句话解释发送方给每个数据包编上序号发出去后启动一个定时器接收方收到后在本地回一个 ACK告诉发送方「这个序号之前的数据我都收到了」如果定时器到期还没等到对应 ACK发送方就把这个包重发一遍。三个机制互相依赖少任何一个都会出问题——没有序号发送方不知道 ACK 对应哪个包没有超时重传丢失的包永远不会补没有 ACK发送方永远不知道对端状态。代码层面第一步是定义包结构。这个结构会同时被发送端、接收端、信道三方引用字段一旦定错后面全是白搭。我一般用一个普通 Python 类配合struct做二进制序列化import struct, zlib class Packet: MAGIC 0x5A TYPE_DATA 1 TYPE_ACK 2 HEADER_SIZE 16 # 2 1 1 4 4 4 def __init__(self, seq0, ack0, ptypeTYPE_DATA, payloadb): self.seq seq self.ack ack self.ptype ptype self.payload payload def to_bytes(self): head struct.pack(!HBBII, self.MAGIC, self.ptype, 0, self.seq, self.ack) checksum zlib.crc32(head self.payload) 0xFFFFFFFF return head struct.pack(!I, checksum) self.payload classmethod def from_bytes(cls, raw): if len(raw) cls.HEADER_SIZE: return None magic, ptype, _, seq, ack struct.unpack(!HBBII, raw[:12]) if magic ! cls.MAGIC: return None checksum struct.unpack(!I, raw[12:16])[0] payload raw[16:] if (zlib.crc32(raw[:12] payload) 0xFFFFFFFF) ! checksum: return None return cls(seqseq, ackack, ptypeptype, payloadpayload)这里用!HBBII指定网络字节序H是 magic占 2 字节两个B分别是包类型和保留字段两个I是 32 位序号和确认号。为什么不用 pickle因为 pickle 的格式随 Python 版本变化而且解析效率低用struct可以把包头固定住也方便以后用别的语言写对端。校验和选择zlib.crc32而不是 md5因为可靠传输只需要发现随机损坏不需要防篡改CRC32 速度快且足够用。from_bytes里所有非法包都返回None调用方统一丢弃这是后面能稳定排查问题的基础。2.2 停等、GBN 与选择重传三种常见做法的取舍基于上面的包格式你可以实现的可靠传输方案有三种对应计算机网络课里的停等协议、Go-Back-NGBN和选择重传SR。三者的核心区别在「窗口」和「重传粒度」。停等是窗口为 1发一个包等确认再发下一个实现最简单但 RTT 一大信道利用率就惨不忍睹。GBN 把窗口扩大允许连续发送多个包接收端只按序接收乱序包直接丢弃发送端超时后从丢失位置开始重传整个窗口。SR 在接收端也开一个窗口乱序包先缓存只重传真正丢失的那一个。三者的取舍可以用一张表说清方案发送窗口接收方处理重传粒度实现成本适用场景停等1顺序接收单个低理解原理、极低带宽链路GBN大于 1乱序丢弃从丢失处重传一批中课程实验、丢包率不高的局域网SR大于 1乱序缓存单个高高丢包、高 RTT 的卫星链路等如果只是完成「基于Python实现的可靠数据传输协议」这个题目GBN 是最常见的做法。原因之一是课程大纲一般默认讲 GBN原因之二是 GBN 只需要维护base和next_seq两个整数调试起来比 SR 容易得多原因之三是本地模拟信道丢包重传一批和重传一个差别不大但代码复杂度低很多。SR 虽然更优雅接收端还要维护接收窗口和重排序缓冲很容易写漏边界。所以后面的实现我按 GBN 展开SR 可以作为扩展。2.3 为什么选 UDP 而不是直接改 TCP场景与边界有人会问TCP 本来就是可靠的直接在 TCP 上写文件不就好了为什么还要造轮子。常见场景有三种。第一是课程实验要求老师想让你理解序列号、确认、超时重传在协议栈里是怎么工作的而 TCP 是内核实现的你没法改它。第二是自定义传输需求实时音视频、游戏同步这类应用数据不太重要时希望用 UDP 加自己的重传策略宁可丢旧包也不能等重传。第三是研究性项目比如想实验一个不同于 TCP 的拥塞控制算法直接在 UDP 之上做更自由。所以这个题目选择 UDP 做承载不是因为 UDP「更好」而是因为它把不可靠这个矛盾露在了外面恰好需要你的协议来兜底。边界也要说清楚UDP 在公网会遇到 NAT 和防火墙不像本地回环测试这么友好你的重传协议如果在真实互联网上跑还要考虑带宽竞争、公平性、路径 MTU。所以课程设计里聪明的做法是「本地模拟信道」而不是直接上公网打流量。真实生产里如果不想从头写常见做法是用 QUIC 这类成熟库或者直接依赖 TCP。自己从零写的协议验证价值大于生产价值。提示丢包模拟是这门实验的灵魂。没有丢包任何协议都能跑得通加了丢包才能看出窗口推进和重传逻辑写没写对。3. 用 Python 从零搭一个 GBN 可靠传输核心类与帧格式设计3.1 发送窗口与序号分配两个指针决定一切GBN 的发送窗口实现实际上只需要两个整数base窗口下沿和next_seq下一个待发序号。发送时如果next_seq - base window就发送序号为next_seq的包然后next_seq 1收到 ACK 时把base移动到ack的位置。窗口内的包都需要保留副本因为超时要重传。用字典做缓存最简单序号是键负载是值class SendBuffer: def __init__(self, window8): self.window window self.buf {} # seq - payload self.base 0 self.next_seq 0 def can_send(self): return self.next_seq - self.base self.window def put(self, seq, payload): self.buf[seq] payload self.next_seq seq 1 def ack(self, ack_seq): for seq in range(self.base, ack_seq): self.buf.pop(seq, None) self.base max(self.base, ack_seq) def timeout_retransmit(self): return list(self.buf.items())说明这里用字典而不是列表是因为序号可能很大未确认包在窗口中不一定连续字典按序号索引更方便。ack()处理的是累积确认ACK 里带的ack_seq表示接收方期望的下一个序号所以base之前的所有包都可以从缓存里删掉。如果收到重复 ACK 且ack_seq basemax会保证窗口不倒退。timeout_retransmit()返回从base开始的所有未确认包正好对应 GBN 一次性重传整个窗口的行为。这个SendBuffer是发送端最核心的单位可以单独写单测验证放进 8 个包、收到 ACK4、确认缓存里只剩下 4 到 7。3.2 发送端主循环超时回调与关闭时序有了发送缓冲发送端就能把文件读进来、按帧切分、逐个发到信道。GBN 的定时器有一个特点整个窗口共用一个定时器而不是每个包一个。我的习惯是只在两个时机重置定时器第一次发送窗口有包时以及收到 ACK 推进base后。代码里常犯的问题是每个包都 new 一个 Timer导致回调乱套。发送端驱动的骨架如下import socket, threading, time class GBNDriver: def __init__(self, sock, dest, window8, rto1.0, max_retry5): self.sock sock self.dest dest self.buffer SendBuffer(window) self.rto rto self.max_retry max_retry self.retry_count 0 self.timer None self.done False def send_file(self, data: bytes, payload_size1024): seq 0 for off in range(0, len(data), payload_size): payload data[off:off payload_size] self.buffer.put(seq, payload) self.sock.sendto(Packet(seqseq, payloadpayload).to_bytes(), self.dest) seq 1 while not self.buffer.can_send(): time.sleep(0.001) self.start_timer_if_needed() self.wait_all_acked() def _ack_listener(self): while not self.done: raw, _ self.sock.recvfrom(65535) pkt Packet.from_bytes(raw) if pkt is None or pkt.ptype ! Packet.TYPE_ACK: continue old_base self.buffer.base self.buffer.ack(pkt.ack) if self.buffer.base old_base: continue self.retry_count 0 self._restart_timer() def _timeout_handler(self): self.retry_count 1 if self.retry_count self.max_retry: self.done True return for seq, payload in self.buffer.timeout_retransmit(): self.sock.sendto(Packet(seqseq, payloadpayload).to_bytes(), self.dest) self._restart_timer()这里我简化了几件事start_timer_if_needed负责在窗口第一次有包时启动一个threading.Timerwait_all_acked用一个带超时的循环等base next_seq。重点说明_ack_listener它必须跑在独立线程里否则发送端主循环发完包后没人收 ACK。收到 ACK 后如果base没变就是重复 ACK直接忽略如果base前进了说明有新包被确认重置retry_count并重启定时器。_timeout_handler里先加重传次数超过max_retry就放弃否则把窗口内所有包重发一遍。这正是 GBN 和 SR 最大的行为差异SR 只重发丢失的序号GBN 重发它后面的所有包。3.3 接收端校验、去重与累积 ACK接收端比发送端简单因为没有窗口管理只有一个期望序号expected。收到数据包后先校验再比对序号相等才写入文件并回 ACK不相等说明是乱序或重复包直接丢弃但要回一个当前期望序号的 ACK让发送端知道「别等了我要的是这个序号」。class GBNReceiver: def __init__(self, sock, output_path): self.sock sock self.expected 0 self.out open(output_path, wb) self.dest None def run_forever(self): while True: raw, addr self.sock.recvfrom(65535) if self.dest is None: self.dest addr pkt Packet.from_bytes(raw) if pkt is None: continue if pkt.ptype Packet.TYPE_DATA: self._handle_data(pkt) def _handle_data(self, pkt): if pkt.seq self.expected: self.out.write(pkt.payload) self.expected 1 self._send_ack(self.expected) else: self._send_ack(self.expected) def _send_ack(self, ack): pkt Packet(seq0, ackack, ptypePacket.TYPE_ACK) self.sock.sendto(pkt.to_bytes(), self.dest)注意run_forever是没有退出条件的课程实验里发送端进程结束后接收端一般由测试脚本 kill 掉。_send_ack里发的 ACK 号是self.expected也就是期望的下一个数据序号发送端收到后把base推到这个位置。乱序数据包在这里被直接丢弃整个接收端不保留任何缓存这是 GBN 接收端的特点。如果要做 SR就得把pkt.seq self.expected的包放进一个缓存列表等缺失的包到了再按顺序交付上层代码量会明显增加。实际调协议时接收端的日志只要记录「收到乱序包 seq多少、期望是多少」就能很快判断发送端的重传是不是在瞎发。4. 在本地跑通最小闭环配置参数与端到端联调4.1 模拟丢包与乱序的信道网络仿真的实现可靠传输的测试离不开信道模拟。常见做法是单写一个channel.py在本地起一个 UDP 中转进程发送端把包发给信道信道按配置丢弃一部分、延迟一部分再转给接收端反向的 ACK 也走同样路径但丢包率可以设小一点比如只丢数据的十分之一。如果不想每来一个包就开线程也可以用queue.Queue做缓冲由单线程统一调度不过最简单的可读做法是先开线程转发。import socket, random, threading, time, argparse class Channel: def __init__(self, data_port, ack_port, recv_port, send_port, loss, delay, jitter): self.data_sock socket.socket(socket.AF_INET, socket.SOCK_DGRAM) self.data_sock.bind((127.0.0.1, data_port)) self.ack_sock socket.socket(socket.AF_INET, socket.SOCK_DGRAM) self.ack_sock.bind((127.0.0.1, ack_port)) self.recv_addr (127.0.0.1, recv_port) self.send_addr (127.0.0.1, send_port) self.loss loss self.delay delay self.jitter jitter def _forward(self, sock, target_addr): raw, _ sock.recvfrom(65535) if random.random() self.loss: return delay max(0, self.delay random.uniform(-self.jitter, self.jitter)) time.sleep(delay) sock.sendto(raw, target_addr) def run(self): while True: threading.Thread( targetself._forward, args(self.data_sock, self.recv_addr), daemonTrue).start() threading.Thread( targetself._forward, args(self.ack_sock, self.send_addr), daemonTrue).start() if __name__ __main__: parser argparse.ArgumentParser() parser.add_argument(--data-port, typeint, default8001) parser.add_argument(--ack-port, typeint, default8003) parser.add_argument(--recv-port, typeint, default8002) parser.add_argument(--send-port, typeint, default8004) parser.add_argument(--loss, typefloat, default0.1) parser.add_argument(--delay, typefloat, default0.05) parser.add_argument(--jitter, typefloat, default0.02) args parser.parse_args() Channel(args.data_port, args.ack_port, args.recv_port, args.send_port, args.loss, args.delay, args.jitter).run()这里用两个监听端口把数据流和 ACK 流分开好处是设置丢包时更灵活数据丢 10%ACK 丢 1%更贴近真实网络。_forward里的time.sleep(delay)会产生随机延迟后到的包可能先被转发这就是乱序的来源。这个实现每收到一个包就起一个新线程只适合做正确性验证如果要把吞吐测准换成 asyncio 协程或线程池更好。参数loss0.1表示 10% 概率丢包delay0.05是 50ms 基础延迟jitter0.02是上下浮动 20ms。把loss设成 1.0就能看到发送端一路重传到max_retry然后退出。4.2 发送端、接收端、信道三者如何协调工作信道起来之后启动顺序很重要。先启动channel.py再启动receiver.py最后启动sender.py。如果先启动发送端它发出的第一批包会丢但这不是大问题发送端会超时重传。为了让启动顺序不成为一种玄学建议在协议里加一个「先发握手包收到 ACK 再发数据」的步骤后面进阶再说。现在先理解最简单的协调方式接收端先就位信道只负责转发发送端重传兜底。本地联调的一组命令如下python channel.py --data-port 8001 --ack-port 8003 --recv-port 8002 --send-port 8004 --loss 0.1 --delay 0.05 python receiver.py --bind-port 8002 --output received.bin python sender.py --local-port 8004 --channel-port 8001 --file test.bin --window 8 --rto 0.5等发送端进程退出后检查received.bin的 md5 是否等于原始文件。这里sender.py --channel-port 8001指的是发往信道的监听端口不是接收端端口。很多第一次做的人会把发送端直接发到8002那样信道就收不到包协议当然跑不通。--rto 0.5是超时 0.5 秒如果channel.py的--delay被调大到了 1 秒rto就必须跟着调大否则信道还没转发完发送端已经超时重传了。出现大量重传时第一反应应该是看参数匹配而不是怀疑代码。线程模型方面发送端和接收端都是「主线程干传输另一个线程收 ACK」信道则是「每个包新起线程」。这三者的 socket 要各自绑定不同的端口否则本地回环上同一个端口是绑定不到的。如果报Address already in use用pkill -f channel.py清掉遗留进程再跑。4.3 参数配置窗口大小、超时重传、最大重传次数怎么设参数表是这份实验最值得抄作业的部分。我一般在一开始就固定一套稳妥参数而不是等翻车了再调参数推荐初值怎么调window8丢包率 10% 时建议 4~8窗口太大会放大 GBN 的重传代价rto0.5s先用RTT * 1.5不知道 RTT 就 0.5频繁重传就加大卡死就减小max_retry5提高到 10 可以更抗丢包但信道全断时等待时间会很长payload_size1024以太网 MTU 1500去掉头部最好不超过 1400避免 IP 分片loss0.1验证可靠性至少要在丢包率 0.1 下跑通再升高到 0.3 观察行为几个经验window并不是越大越好。GBN 在丢包率较高时窗口越大一次超时重传的包越多反而放大浪费。本地测试经典组合是window8, loss0.1, rto0.5跑通后再分别拧参数观察吞吐和重传次数变化。max_retry不要设成无限否则信道完全断开时发送端不会退出程序会挂在那里像死锁。最后一点如果用的是 Python 3.8argparse解析出来的参数默认是字符串记得在add_argument里写typeint或typefloat不然window会被当成字符串跟整数比较报错信息很不直观。5. 避坑指南Python 实现可靠传输最容易翻车的 5 个地方5.1 定时器与线程RTO 被谁吃了最经典的翻车现场发送端不停地重传哪怕信道延迟只有几毫秒。拆开原因多半是threading.Timer没有管理好。很多人每个包都Timer(rto, timeout)结果旧 Timer 还在新 Timer 又起来回调里重传了整个窗口窗口越堵越乱。更隐蔽的是 Python 3.8 中Timer.cancel()只保证回调不再从线程池里取出来执行但如果你已经把回调交给线程准备执行它还是可能跑起来。解决方法是给定时器加一个版本号def _restart_timer(self): self.timer_gen 1 gen self.timer_gen if self.timer is not None: self.timer.cancel() t threading.Timer(self.rto, self._timeout_guard, args(gen,)) t.daemon True self.timer t t.start() def _timeout_guard(self, gen): if gen ! self.timer_gen: return self._timeout_handler()每次重启定时器都把gen加 1旧回调即使执行也会被挡回去。这个技巧比单纯cancel可靠得多是我调 GBN 时最后悔没早点加上的代码。5.2 UDP 也是会「粘包」的缓冲区背后的损失严格说 UDP 有报文边界不会像 TCP 那样粘包半包但实际编码时经常会遇到类似现象接收端一次读上来的数据长度不对解析头部就失败。原因通常是发送端把多个包拼到一个bytearray里一次sendto这是不对的UDP 会把它当成一个报文接收端按一个包解析自然失败。另一个坑是操作系统的 UDP 接收缓冲区太小发送端一开窗口猛发接收端的 socket 来不及读包在操作系统里被直接丢弃现象就是「明明没开丢包协议却疯狂重传」。解决接收端 socket 显式调大SO_RCVBUF到 10MB并对每个报文做len(raw) Packet.HEADER_SIZE的检查非法包直接丢弃而不是抛异常把线程打死。还有一个小细节用recvfrom而不是recv否则拿不到对端地址ACK 无处可回。5.3 重复 ACK 与快速重传冲突如果你熟悉 TCP 调优可能会想「收到三个重复 ACK 就重传」但这个习惯照搬到 GBN 上是会出大事的。现象丢包率才 2%重传数量却占了一半流量。原因GBN 的接收端对每个乱序包都会回「期望序号」的 ACK发送端会收到大量重复 ACK如果按三个重复 ACK 就重传那个丢失的包还没等到超时整个窗口就重传了后续 ACK 又重复形成雪崩。解决在 GBN 里放弃快速重传只认超时。收到重复 ACK 时直接忽略因为base不变。越早克制住「重传强迫症」GBN 跑得越稳。如果以后实现 SR再单独统计重复 ACK 数量触发快速重传那是另一套逻辑两套机制不能混着写。5.4 关闭连接时的数据丢失最后一个包永远传不过去可靠性测试最容易忽略的就是关闭阶段。现象文件校验和总是差最后几个字节但重传日志显示所有数据都 ACK 了。原因发送端发完数据base next_seq定时器取消了但这个时候接收端可能还在写文件发送端如果直接close()socket 会把还在队列里的数据丢弃接收端还没发出最后一个 ACK这个 ACK 就没法送达。更常见的是发送端发完所有数据就退出接收端run_forever()还在等但发送端进程没了这不算丢数据是你没实现关闭握手。解决显式设计一个 FIN 过程——发送端发完数据后发一个TYPE_FIN包接收端回FIN_ACK并关闭文件发送端收到FIN_ACK后才退出。在最小实验里也可以偷懒让发送端等base next_seq后再sleep(0.5)再退出给最后一个 ACK 留时间。这个方法不优雅但真能救命是后悔药级别的经验。5.5 Python 性能黑洞别让小包拖死协议现象本地回环 UDP 传 100MB 文件吞吐比 TCP 慢了一个数量级。原因不只是 Python 慢而是每个包的处理链很长——构造对象、CRC32、系统调用、线程切换、日志打印每一步都在放大开销。当你把payload_size设成 256 字节4 万个包就是 4 万次 socket sendto 和线程调度。解决思路有三个第一payload_size往 1400 字节附近提减少包数量第二用zlib.crc32而不用hashlib.md5CRC32 有硬件加速一次算 1KB 数据基本不耗时间第三日志不要放在发送热路径里每 N 个包统计一次。我见过最夸张的坑是有人为了调试在sendto前打印整个包的内容这个热度比协议本身还高关掉日志后吞吐立刻翻倍。6. 把可靠传输做成可验证的东西校验、日志与性能验证的实战技巧6.1 用丢包率矩阵验证协议边界不要只测一次就宣布成功。常见做法是写一个循环对loss0, 0.05, 0.1, 0.2, 0.5分别跑一遍每次结束后用md5sum对比发送端和接收端文件全部一致才算过。shell 脚本越简单越好for loss in 0 0.05 0.1 0.2 0.5; do python channel.py --loss $loss --delay 0.03 --data-port 8101 --ack-port 8103 --recv-port 8102 --send-port 8104 CHANNEL_PID$! python receiver.py --bind-port 8102 --output /tmp/recv_$loss.bin RECV_PID$! sleep 0.2 python sender.py --local-port 8104 --channel-port 8101 --file test.bin sleep 0.5 kill $RECV_PID $CHANNEL_PID 2/dev/null md5sum test.bin /tmp/recv_$loss.bin done看输出里两个 md5 值是否一样或者直接diff。我一般还会在同一轮把window从 4 改到 16看在高丢包下窗口变大是否重传飙升。这个矩阵测完协议基本可以放心交。6.2 让黑匣子变透明结构化日志与状态快照协议跑不通时最烦的是不知道到底卡在哪。用print打日志在多线程下会交错根本没法看常见做法是换logging并在关键点记录seq、ack、event、retry。我习惯在发送端每次重传时打一条在接收端每次收到乱序包时打一条这样一复盘就能定位问题。示例import logging def log_retransmit(self, seq): logging.info( eventretransmit seq%d window[%d,%d) retry%d, seq, self.buffer.base, self.buffer.next_seq, self.retry_count, )这里window[%d,%d)打印窗口范围能看出base有没有推进retry_count能看出是否接近max_retry。平时默认WARNING查问题用INFO不要在热路径里打完整 payload。6.3 性能验证的一个小技巧控变量比大小不管调窗口还是调 payload都要保证除了被测参数其他条件不变。比如拿同样的文件、同样的丢包率只改window记录总耗时和重传次数。我最后悔的事就是用「感觉」调参结果把rto和delay一起改了不知道是谁的功劳。后来固定一个基准配置改之前先跑一遍基线再改再对比才不再靠玄学。我在写这类协议时最大的教训是能跑通的代码和可靠的代码之间差着一整套丢包矩阵和日志复盘。第一次跑通时我高兴得太早关掉丢包重测后一切正常以为稳了结果在 5% 丢包下直接翻车从那以后每次改完协议我都会先跑一遍上面的矩阵再提交。这个习惯算不上聪明但真的能救命希望帮到你。本文还有配套的精品资源点击获取