滑动窗口协议深度解析：GBN与SR的机制、实现与选型指南

1. 为什么滑动窗口是可靠传输的命门

搞过网络协议栈的人都有一个共识：可靠传输这件事，说到底就是在“效率”和“正确性”之间找平衡。而滑动窗口机制，就是那个把平衡点捏在手里的核心设计。你如果只背过GBN和SR的定义，大概率会在实际调优或者面试深挖的时候卡壳——因为真正难的不是记住“回退N步”和“选择重传”这两个名字，而是搞清楚它们各自在什么场景下会出问题、窗口该怎么设、缓冲区该怎么管。

这篇文章面向的是正在学计算机网络、准备面试、或者在实际项目中需要手写可靠传输逻辑的开发者。我会把GBN和SR从设计动机到实操细节全部拆开，包括窗口计算、序号空间分配、缓冲区管理、异常处理，以及我在实际模拟实现中踩过的坑。读完你至少能做到两件事：第一，能手推两种协议的窗口边界和序号回绕条件；第二，能根据具体链路特征判断该选哪种方案。

滑动窗口的本质是什么？你可以把它想象成一个传送带。发送方把数据包一个个放上去，接收方从另一头取。传送带的长度就是窗口大小，它决定了在等待确认的这段时间里，最多能有多少数据在“路上”。如果传送带太短，链路带宽利用率上不去；如果太长，序号空间和缓冲区又扛不住。GBN和SR的区别，就在于当传送带上某个包裹丢了的时候，是整段重传还是只补那一个。

注意：很多人学的时候把GBN和SR当成纯理论，觉得实际网络里都是TCP那种混合机制。但恰恰是这两种极端模型，能帮你把TCP的很多设计选择看得更清楚。比如TCP为什么用累积确认？为什么SACK选项那么重要？根子都在GBN和SR的取舍里。

2. GBN与SR的核心机制拆解

2.1 GBN的工作逻辑与累积确认

GBN的全称是Go-Back-N，中文叫回退N步。它的核心规则非常粗暴：发送方可以连续发送多个帧而不等待确认，但接收方只按序接收，一旦某个帧丢失或出错，后面所有帧全部丢弃，发送方从丢失的那个帧开始重传所有后续帧。

这里的关键设计是累积确认。接收方不需要对每个帧单独发ACK，而是对最后一个按序到达的帧发确认，表示“这个帧以及它之前的所有帧我都收到了”。比如接收方收到了0、1、2、3，它只需要发ACK3，发送方就知道0到3全部安全到达。

发送方的窗口结构是这样的：窗口大小为N，序号空间必须满足2^k >= N+1，其中k是序号位数。为什么是N+1而不是N？因为如果序号空间等于N，当窗口滑动后，新旧序号会重叠，接收方无法区分一个帧是新的还是重传的。举个例子，N=4，序号用2位表示（0到3），发送方发了0、1、2、3，接收方全部确认。但ACK全部丢失，发送方超时重传0。此时接收方看到序号0，它无法判断这是新的一轮数据还是上一轮的重复帧。所以序号空间必须比窗口大至少1。

GBN的发送方需要维护三个关键变量：

• base：窗口基序号，指向最早未确认的帧
• nextseqnum：下一个要发送的序号
• N：窗口大小

发送方的操作逻辑用伪代码表示：

# GBN 发送方核心逻辑 base = 1 nextseqnum = 1 N = 4 # 窗口大小 def send_data(): global base, nextseqnum if nextseqnum < base + N: # 窗口未满，可以发送 send_frame(nextseqnum) if base == nextseqnum: start_timer() # 只为最早的未确认帧启动定时器 nextseqnum += 1 else: # 窗口已满，拒绝发送，等待ACK refuse_data() def on_ack_received(ack_num): global base # 累积确认：ack_num表示ack_num及之前全部收到 base = ack_num + 1 if base == nextseqnum: stop_timer() else: restart_timer() # 还有未确认帧，重启定时器 def on_timeout(): # 超时：回退到base，重传所有已发送但未确认的帧 for i in range(base, nextseqnum): send_frame(i) restart_timer()

接收方的逻辑更简单：维护一个期望序号expectedseqnum，初始为1。收到帧后检查序号：

• 如果序号等于expectedseqnum，接收并交付，然后expectedseqnum += 1，发送ACK
• 如果序号不等于expectedseqnum，直接丢弃，并发送对最后一个按序到达帧的ACK（重复ACK）

这种设计的优点是接收方缓冲区只需要一个帧的空间，实现极其简单。但代价也很明显：一个帧出错，后面所有正确到达的帧都要被丢弃并重传。如果窗口很大、误码率又高，GBN的效率会急剧下降。

2.2 SR的工作逻辑与独立确认

SR的全称是Selective Repeat，中文叫选择重传。它的核心改进是：接收方可以接收乱序到达的帧，先缓存起来，只要求发送方重传真正丢失的那些帧。

SR的发送方需要为每个已发送但未确认的帧单独维护一个定时器。当某个帧超时，只重传那一个帧，而不是像GBN那样回退重传一批。接收方则需要一个接收窗口，窗口内的帧无论顺序如何都先收下，然后按序交付给上层。

SR的窗口大小约束和GBN不同：窗口大小N必须满足2N <= 2^k，即N <= 2^(k-1)。为什么？因为发送窗口和接收窗口在序号空间上不能重叠。如果N太大，发送方的新帧和接收方期望的旧帧可能共用同一个序号，导致接收方无法区分。比如k=3（序号0到7），如果N=5，发送窗口可能是0到4，接收窗口可能是4到0（回绕），序号4同时出现在两个窗口里，接收方收到序号4时无法判断这是新帧还是重传帧。

SR发送方的核心变量：

• send_base：发送窗口基序号
• nextseqnum：下一个待发送序号
• N：窗口大小
• 每个已发送帧的定时器

# SR 发送方核心逻辑 send_base = 1 nextseqnum = 1 N = 4 acked = [False] * (N + 1) # 标记窗口内哪些帧已确认 def send_data(): global nextseqnum if nextseqnum < send_base + N: send_frame(nextseqnum) start_timer(nextseqnum) # 每个帧独立定时器 nextseqnum += 1 else: refuse_data() def on_ack_received(ack_num): global send_base if send_base <= ack_num < send_base + N: acked[ack_num - send_base] = True stop_timer(ack_num) # 滑动窗口：如果基序号已确认，向前移动 while acked[0]: acked.pop(0) acked.append(False) send_base += 1 def on_timeout(seq_num): # 只重传超时的那个帧 send_frame(seq_num) restart_timer(seq_num)

SR接收方的逻辑：

• 维护接收窗口rcv_base到rcv_base + N - 1
• 收到帧后，如果序号在接收窗口内，缓存该帧并发送ACK
• 如果序号等于rcv_base，将该帧及后续连续已缓存的帧一起交付上层，然后滑动窗口
• 如果序号在窗口外（比如已经确认过的旧帧），仍然发送ACK，但不缓存

# SR 接收方核心逻辑 rcv_base = 1 N = 4 buffer = {} # 缓存乱序到达的帧 def on_frame_received(seq_num, data): global rcv_base if rcv_base <= seq_num < rcv_base + N: # 在接收窗口内 buffer[seq_num] = data send_ack(seq_num) # 按序交付 while rcv_base in buffer: deliver_to_app(buffer[rcv_base]) del buffer[rcv_base] rcv_base += 1 elif seq_num < rcv_base: # 旧帧，可能ACK丢失导致的重传，重新确认 send_ack(seq_num) # 窗口外的帧直接忽略

SR的优点是重传效率高，尤其在高带宽、高延迟、误码率不低的环境下，优势非常明显。但代价是接收方需要N个帧的缓冲区，发送方需要N个定时器，实现复杂度显著上升。

2.3 两种协议的窗口与序号空间对比

这张表建议你面试前直接背下来。但更重要的是理解每一行背后的“为什么”。比如为什么GBN只需要一个定时器？因为它只关心最早的未确认帧，只要那个帧超时了，后面所有帧都要重传，所以没必要给每个帧单独计时。而SR必须给每个帧单独计时，因为它只重传超时的那个，如果不单独计时，就无法知道具体哪个帧丢了。

3. 实操过程：从零实现一个滑动窗口模拟器

3.1 环境准备与基础框架搭建

我用的环境是Python 3.10，不需要任何第三方库，标准库的threading、queue、time就够了。为什么选Python？因为它的语法足够简洁，能让你把精力集中在协议逻辑本身，而不是内存管理和类型声明上。如果你要写生产级的协议栈，那肯定用C或者Rust，但做教学模拟和逻辑验证，Python是最快出结果的。

先搭一个基础框架，把发送方、接收方、信道三个角色分开：

import threading import queue import time import random class Channel: """模拟不可靠信道：可能丢包、可能出错、可能乱序""" def __init__(self, loss_rate=0.1, corrupt_rate=0.05): self.loss_rate = loss_rate self.corrupt_rate = corrupt_rate self.sender_to_receiver = queue.Queue() self.receiver_to_sender = queue.Queue() def transmit(self, packet, direction): """direction: 's2r' 或 'r2s'""" if random.random() < self.loss_rate: return # 丢包 if random.random() < self.corrupt_rate: packet = corrupt(packet) # 模拟比特错误 if direction == 's2r': self.sender_to_receiver.put(packet) else: self.receiver_to_sender.put(packet)

这个信道模型有三个关键参数：loss_rate控制丢包概率，corrupt_rate控制比特错误概率，队列本身不保证顺序（虽然Python的Queue是FIFO，但你可以通过随机延迟来模拟乱序）。我建议一开始把丢包率设成0.1，这样能快速看到重传行为，调试通了再降到0.01看性能差异。

提示：不要一上来就设成0丢包率，那样你根本测不出协议的正确性。我见过有人写完GBN，在无丢包环境下跑通了就以为对了，结果一加丢包就死锁。必须从一开始就在有丢包的环境下测试。

3.2 GBN发送方与接收方的完整实现

先定义帧的结构。一个帧至少包含序号、数据、校验和（用来模拟差错检测）：

class Frame: def __init__(self, seq_num, data, is_ack=False): self.seq_num = seq_num self.data = data self.is_ack = is_ack self.checksum = self.compute_checksum() def compute_checksum(self): # 简单校验和：数据字节和模256 return sum(self.data.encode()) % 256 if isinstance(self.data, str) else 0 def is_valid(self): return self.checksum == self.compute_checksum()

GBN发送方的完整实现：

class GBNSender: def __init__(self, channel, window_size=4, timeout=1.0): self.channel = channel self.N = window_size self.base = 0 self.nextseqnum = 0 self.timeout = timeout self.timer = None self.lock = threading.Lock() self.total_sent = 0 self.total_retrans = 0 def send(self, data_list): """发送一批数据""" for data in data_list: self._send_data(data) # 等待所有确认 while self.base < len(data_list): time.sleep(0.01) def _send_data(self, data): with self.lock: if self.nextseqnum < self.base + self.N: frame = Frame(self.nextseqnum, data) self.channel.transmit(frame, 's2r') self.total_sent += 1 if self.base == self.nextseqnum: self._start_timer() self.nextseqnum += 1 else: # 窗口满，等待 time.sleep(0.05) self._send_data(data) # 递归重试 def _start_timer(self): if self.timer: self.timer.cancel() self.timer = threading.Timer(self.timeout, self._on_timeout) self.timer.start() def _on_timeout(self): with self.lock: # 回退：重传base到nextseqnum-1的所有帧 for seq in range(self.base, self.nextseqnum): frame = Frame(seq, f"data_{seq}") self.channel.transmit(frame, 's2r') self.total_retrans += 1 self._start_timer() def receive_ack(self, ack_num): with self.lock: if ack_num >= self.base: self.base = ack_num + 1 if self.base == self.nextseqnum: if self.timer: self.timer.cancel() self.timer = None else: self._start_timer()

GBN接收方的实现：

class GBNReceiver: def __init__(self, channel): self.channel = channel self.expectedseqnum = 0 self.received_data = [] def run(self): while True: try: frame = self.channel.sender_to_receiver.get(timeout=0.1) if not frame.is_valid(): # 帧出错，发送对最后一个按序帧的ACK self._send_ack(self.expectedseqnum - 1) continue if frame.seq_num == self.expectedseqnum: self.received_data.append(frame.data) self._send_ack(self.expectedseqnum) self.expectedseqnum += 1 else: # 乱序帧，丢弃，重复ACK self._send_ack(self.expectedseqnum - 1) except queue.Empty: continue def _send_ack(self, ack_num): ack = Frame(ack_num, "", is_ack=True) self.channel.transmit(ack, 'r2s')

跑起来之后你会发现，GBN在丢包率10%的情况下，重传次数可能达到总发送次数的30%到50%。因为一个帧丢了，后面窗口内所有帧都要重传。你可以把total_retrans / total_sent打印出来，直观感受一下。

3.3 SR发送方与接收方的完整实现

SR的发送方需要为每个帧维护独立定时器，我用一个字典来管理：

class SRSender: def __init__(self, channel, window_size=4, timeout=1.0): self.channel = channel self.N = window_size self.send_base = 0 self.nextseqnum = 0 self.timeout = timeout self.timers = {} # seq_num -> Timer self.acked = {} # seq_num -> bool self.lock = threading.Lock() self.total_sent = 0 self.total_retrans = 0 def send(self, data_list): for data in data_list: self._send_data(data) while self.send_base < len(data_list): time.sleep(0.01) def _send_data(self, data): with self.lock: if self.nextseqnum < self.send_base + self.N: seq = self.nextseqnum frame = Frame(seq, data) self.channel.transmit(frame, 's2r') self.total_sent += 1 self.acked[seq] = False self._start_timer(seq) self.nextseqnum += 1 else: time.sleep(0.05) self._send_data(data) def _start_timer(self, seq): if seq in self.timers: self.timers[seq].cancel() timer = threading.Timer(self.timeout, self._on_timeout, args=[seq]) self.timers[seq] = timer timer.start() def _on_timeout(self, seq): with self.lock: if seq in self.acked and not self.acked[seq]: # 只重传这一个帧 frame = Frame(seq, f"data_{seq}") self.channel.transmit(frame, 's2r') self.total_retrans += 1 self._start_timer(seq) def receive_ack(self, ack_num): with self.lock: if ack_num in self.acked and not self.acked[ack_num]: self.acked[ack_num] = True if ack_num in self.timers: self.timers[ack_num].cancel() del self.timers[ack_num] # 滑动窗口 while self.send_base in self.acked and self.acked[self.send_base]: del self.acked[self.send_base] self.send_base += 1

SR接收方的实现：

class SRReceiver: def __init__(self, channel, window_size=4): self.channel = channel self.N = window_size self.rcv_base = 0 self.buffer = {} self.received_data = [] def run(self): while True: try: frame = self.channel.sender_to_receiver.get(timeout=0.1) if not frame.is_valid(): continue seq = frame.seq_num if self.rcv_base <= seq < self.rcv_base + self.N: # 在接收窗口内 if seq not in self.buffer: self.buffer[seq] = frame.data self._send_ack(seq) # 按序交付 while self.rcv_base in self.buffer: self.received_data.append(self.buffer[self.rcv_base]) del self.buffer[self.rcv_base] self.rcv_base += 1 elif seq < self.rcv_base: # 旧帧，重新确认 self._send_ack(seq) except queue.Empty: continue def _send_ack(self, ack_num): ack = Frame(ack_num, "", is_ack=True) self.channel.transmit(ack, 'r2s')

跑对比测试的时候，把两个协议的total_retrans / total_sent打印出来，在同样的丢包率下，SR的重传率通常只有GBN的三分之一到五分之一。但SR的代码量大概是GBN的两倍，而且定时器管理更容易出bug。

3.4 窗口大小与序号位数的计算实操

这部分是很多人容易翻车的地方。假设序号字段用3位，也就是序号范围0到7，那么：

GBN的窗口大小计算：

• 约束条件：2^k >= N + 1
• k=3，所以8 >= N + 1，即N <= 7
• 最大窗口是7

SR的窗口大小计算：

• 约束条件：2^k >= 2N
• k=3，所以8 >= 2N，即N <= 4
• 最大窗口是4

你可以自己写个脚本验证一下，当N超过最大值时会发生什么。比如GBN取N=8，序号0到7，发送方发0到7，接收方全部确认但ACK全丢。发送方超时重传0，接收方此时期望的序号也是0（因为它以为0到7都收到了，下一轮从0开始），它会把重传的0当成新帧收下，数据就乱了。

def validate_window_size(k, N, protocol): """验证窗口大小是否合法""" seq_space = 2 ** k if protocol == 'GBN': if N + 1 <= seq_space: return True, f"合法：N={N}, 序号空间={seq_space}, 需要{N+1}" else: return False, f"非法：N={N}, 序号空间={seq_space}, 需要{N+1}" elif protocol == 'SR': if 2 * N <= seq_space: return True, f"合法：N={N}, 序号空间={seq_space}, 需要{2*N}" else: return False, f"非法：N={N}, 序号空间={seq_space}, 需要{2*N}" # 测试 print(validate_window_size(3, 7, 'GBN')) # (True, '合法...') print(validate_window_size(3, 8, 'GBN')) # (False, '非法...') print(validate_window_size(3, 4, 'SR')) # (True, '合法...') print(validate_window_size(3, 5, 'SR')) # (False, '非法...')

注意：实际协议设计中，序号位数往往是固定的（比如TCP用32位），窗口大小是根据链路带宽延迟积动态调整的。但窗口大小的上限永远受序号空间约束。TCP的窗口字段是16位，但通过窗口缩放选项可以扩大到30位，这就是为了在长肥管道上提高吞吐量。

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

4.1 定时器管理中的死锁与竞态

我在实现SR的时候踩过最大的坑就是定时器竞态。因为每个帧一个定时器，当ACK到达和定时器超时同时发生时，如果不加锁，会出现“ACK已经处理了，但定时器还是触发了重传”的情况。更糟糕的是，如果重传的帧又触发了新的定时器，而旧的定时器没有正确取消，会导致同一个帧被无限重传。

解决方案是在定时器回调里检查该帧是否已经被确认：

def _on_timeout(self, seq): with self.lock: if seq in self.acked and not self.acked[seq]: # 只有未确认才重传 self.channel.transmit(Frame(seq, f"data_{seq}"), 's2r') self._start_timer(seq) # 如果已确认，什么都不做，定时器自然消亡

另一个坑是GBN的定时器重启。当收到一个累积ACK但窗口还没空时，必须重启定时器。我一开始忘了这一步，导致发送方在收到部分确认后，定时器在旧的时间点触发，把已经确认的帧又重传了一遍。

4.2 序号回绕导致的接收方误判

序号回绕是滑动窗口协议里最隐蔽的bug。假设序号空间是8（0到7），窗口大小是7（GBN最大）。发送方发了0到6，接收方全部收到并确认。但ACK全部丢失，发送方超时重传0。接收方此时期望的序号是7，它收到0后，因为0不在期望范围内，会丢弃并重复ACK6。发送方收到ACK6后，base变成7，然后发送7。接收方收到7，交付，期望序号回绕到0。此时发送方再发0，接收方正常接收。

但如果窗口大小是8（非法），发送方发了0到7，接收方全部收到，期望序号回绕到0。ACK全丢，发送方重传0。接收方看到0，以为是新帧，收下并交付。数据就重复了。

排查这种问题的方法很简单：在接收方打印每次收到的序号和当前期望序号，在发送方打印每次发送的序号和窗口范围。一旦发现接收方交付了重复数据，或者发送方的窗口滑动异常，基本就是序号空间不够。

4.3 信道丢包率对两种协议性能的影响对比

我跑了一组对比实验，数据量1000个帧，窗口大小4，超时1秒，序号3位。结果如下：

从数据可以看出，在低丢包率下，两者差距不大，GBN甚至因为实现简单而略快。但丢包率超过5%之后，SR的优势就非常明显了。丢包率20%时，GBN的重传次数是SR的4倍多，总耗时是SR的2.6倍。

这个实验说明一个道理：没有绝对优劣的协议，只有适合特定链路的协议。如果链路质量很好（比如光纤），GBN的简单性反而更有优势；如果链路质量差（比如无线），SR的重传效率能救命。

4.4 常见问题速查表

提示：实际调试时，我建议在发送方和接收方各加一个日志文件，记录每个帧的发送、接收、确认、超时事件。事后用脚本分析日志，比在控制台看滚动输出高效得多。

5. 从理论到实战的选型建议

5.1 什么场景该用GBN，什么场景该用SR

GBN适合的场景：链路误码率极低（比如有线局域网）、接收方资源极度受限（比如嵌入式传感器）、协议实现要尽可能简单（比如教学演示）。我做过一个工业控制器的项目，用的就是类似GBN的机制，因为那个场景下链路非常稳定，而且控制器的RAM只有几KB，根本放不下SR的接收缓冲区。

SR适合的场景：无线链路、卫星链路、任何误码率超过1%的环境。移动通信里的RLC层（无线链路控制）用的就是SR的变种，因为无线信道的误码率天然就高，GBN的重传放大效应会导致吞吐量崩溃。

还有一个折中方案叫选择性确认（SACK），它本质上是GBN的累积确认加上SR的选择性重传。TCP的SACK选项就是干这个的：接收方通过SACK告诉发送方“我收到了哪些乱序的块”，发送方只重传缺失的块。这样既保留了累积确认的简洁性，又获得了SR的重传效率。

5.2 窗口大小设置的工程经验

窗口大小不是越大越好。窗口太大，接收方缓冲区压力大，序号空间要求高，而且一旦出错重传的数据量也大。窗口太小，带宽利用率上不去。

工程上的经验公式是：窗口大小 >= 带宽延迟积（BDP）。BDP = 链路带宽 × 往返时间。比如带宽100Mbps，RTT 10ms，BDP = 100Mbps × 10ms = 1Mbit = 125KB。如果帧大小是1500字节，那么窗口至少需要125000 / 1500 ≈ 84个帧。

但窗口大小还受序号空间约束。如果序号是8位（256个序号），SR的窗口最大是128。如果BDP算出来需要200个帧的窗口，那要么增大序号位数，要么接受带宽利用率下降。

我在实际项目中的做法是：先算BDP，然后取窗口 = min(BDP对应的帧数, 序号空间允许的最大值)。如果两者差距太大，就考虑增大序号字段或者用窗口缩放。

5.3 协议演进中的滑动窗口思想

滑动窗口的思想不只用在数据链路层和传输层。应用层的消息队列（比如Kafka的消费者偏移量）、分布式系统的共识协议（比如Raft的日志复制），本质上都在用滑动窗口做流量控制和可靠传输。

Kafka的消费者用offset来标记消费进度，broker只保留最近一段时间的消息，消费者可以重置offset来重放数据。这其实就是GBN的累积确认思想：offset之前的消息全部确认，offset之后的消息可以并行消费。

Raft的日志复制更像SR：leader为每个follower维护一个nextIndex，只重传follower缺失的日志条目，而不是回退重传所有。这种设计让Raft在 follower 短暂失联后能快速追上，而不是从头同步。

理解GBN和SR的取舍，能帮你更快地看懂这些上层协议的设计动机。你会发现，可靠传输的核心矛盾永远是“重传效率”和“实现复杂度”之间的权衡，而滑动窗口只是解决这个矛盾的一种通用框架。

5.4 我个人的实操体会

最后分享几个我在实现和调试滑动窗口协议时总结的小技巧。第一，永远先用小序号空间测试。用3位序号、窗口大小4来跑，这样序号回绕很快，能迅速暴露边界bug。如果直接用32位序号，你可能跑一整天都碰不到回绕问题。第二，在接收方加一个重复帧检测器。如果同一个序号被交付了两次，立刻报警并打印上下文。这个检测器帮我抓到过至少三个隐蔽的竞态bug。第三，不要迷信理论最大窗口。理论最大窗口是2^(k-1)（SR）或2^k - 1（GBN），但实际用的时候建议留一点余量，比如取理论值的80%，这样能容忍一些实现上的边界误差。

这个模拟器后续还可以扩展的方向：加入动态窗口调整（根据RTT和丢包率自动调窗口）、加入SACK选项、模拟TCP的拥塞控制。如果你把GBN和SR都手写一遍，再去看TCP的源码，会发现很多设计选择都变得理所当然。