5.2 KCP 从零实现
5.1 证明了队头阻塞的真实代价，这一节把替代方案写出来：一个教学版 KCP。它不追求与原版 skywind3000/kcp 逐行对齐，而是把「可靠 UDP 到底要哪几个零件」讲透。相对原版做了三处精简：去掉拥塞控制（没有 cwnd/ssthresh，速率只由窗口与重传节奏决定）、去掉会话协商与 probe 计数（conv 由上层指定）、Recv 按整条消息交付（原版按分片交付，应用自己拼）；保留最小可用集：ARQ、滑动窗口、快速重传、RTO 估计。
报文头与三个命令
报文头固定 24 字节，字段顺序与 KCP 一致：conv(4) cmd(1) frg(1) wnd(2) ts(4) sn(4) una(4) len(4)，依次是连接号、命令、分片、窗口、时间戳、序号、累积确认与载荷长度；教学版把 MTU 固定成 512，于是 mss = 512 - 24 = 488。三个命令是 PUSH(81)、ACK(82)、WASK(83)：PUSH 带载荷，ACK 回执并回显时间戳（供对端算 RTT），WASK 是对端窗口为 0 时的探测。frg 是分片计数，一条消息切成 N 片时末片 frg=0，接收端据此判断消息边界；una 是「已连续收到的最大序号 +1」，即累积确认。下面只贴 segment 结构，线格式的编解码在同目录 segment.go：
// snippet: internal/kcp/segment.go 的 segment 结构
type segment struct {
conv uint32
cmd, frg uint8
wnd uint16
ts, sn, una uint32
data []byte
xmit, rto uint32 // 已发送次数、本段当前的重传超时
resendts uint32 // 下次重传时刻
fastack uint32 // 被后续 ACK 跳过的次数
}
状态机：Send / Recv / Input / Update
对外接口只有四个；internal/kcp/kcp.go 是本机跑绿的那一份，逐字贴出：
// file: internal/kcp/kcp.go
// tier: L1
// verify: go test ./internal/kcp/
package kcp
import "sort"
type outputFunc func(buf []byte)
type ackItem struct{ sn, ts uint32 }
// KCP 是教学版可靠 UDP：有序、可靠，但重传节奏与窗口由自己掌控，
// 因此可以针对弱网调参，而不像内核 TCP 那样只能接受固定的拥塞控制。
type KCP struct {
conv uint32
output outputFunc
now, tsFlush uint32
sndUna, sndNxt uint32 // 最早未确认 / 下一个待分配的序号
sndWnd, rmtWnd uint32 // 本端在途上限 / 对端通告的接收窗口
rcvWnd, rcvNxt uint32
sndQueue []*segment // 尚未分配序号
sndBuf []*segment // 已发送、等待确认
rcvBuf []*segment // 已到达、等待补齐
rcvQueue []*segment // 已按序、等待应用读取
srtt, rttvar int32
rto int32
interval uint32
fastresend int
acklist []ackItem
retransmits uint32
}
// NewKCP 建立一个连接；output 是底层发包回调，通常写到 UDP socket。
func NewKCP(conv uint32, output outputFunc) *KCP {
return &KCP{conv: conv, output: output, sndWnd: defaultSendWnd, rmtWnd: defaultRecvWnd,
rcvWnd: defaultRecvWnd, rto: defaultRTO, interval: defaultInterval, fastresend: defaultFastResend}
}
// SetWindow 设置本端发送窗口与接收窗口；RTO/Retransmits 供测试观测内部状态。
func (k *KCP) SetWindow(snd, rcv uint32) { k.sndWnd, k.rcvWnd = snd, rcv }
func (k *KCP) RTO() int32 { return k.rto }
func (k *KCP) Retransmits() uint32 { return k.retransmits }
// Send 把消息放进发送队列；超过 mss 时按 mss 分片，末片 frg=0。
func (k *KCP) Send(data []byte) int {
if len(data) == 0 {
return -1
}
count := (len(data) + mss - 1) / mss
if count > 255 {
return -1
}
for i := 0; i < count; i++ {
size := mss
if len(data) < size {
size = len(data)
}
seg := &segment{conv: k.conv, cmd: cmdPush, frg: uint8(count - i - 1), wnd: uint16(k.rcvWnd)}
seg.data = append([]byte(nil), data[:size]...)
data = data[size:]
k.sndQueue = append(k.sndQueue, seg)
}
return 0
}
// Recv 取出一条完整消息；-1 表示暂无完整消息，-2 表示缓冲区太小。
func (k *KCP) Recv(buf []byte) int {
need, last := 0, -1
for i, seg := range k.rcvQueue {
need += len(seg.data)
if seg.frg == 0 {
last = i
break
}
}
if last < 0 {
return -1
}
if need > len(buf) {
return -2
}
copied := 0
for _, seg := range k.rcvQueue[:last+1] {
copied += copy(buf[copied:], seg.data)
}
k.rcvQueue = k.rcvQueue[last+1:]
return copied
}
// Input 处理一个收到的报文，通常是 UDP 收包后直接喂进来。
func (k *KCP) Input(data []byte) int {
seg, err := decodeSegment(data)
if err != nil || seg.conv != k.conv {
return -1
}
if seg.wnd > 0 {
k.rmtWnd = uint32(seg.wnd)
} else {
k.rmtWnd = 1
}
k.parseUna(seg.una)
k.shrinkBuf()
switch seg.cmd {
case cmdAck:
if itimediff(k.now, seg.ts) >= 0 {
k.updateRTT(itimediff(k.now, seg.ts))
}
k.parseAck(seg.sn)
k.shrinkBuf()
case cmdPush:
if itimediff(seg.sn, k.rcvNxt+k.rcvWnd) < 0 {
k.acklist = append(k.acklist, ackItem{seg.sn, seg.ts})
k.parseData(seg)
}
case cmdWask: // 对端探测窗口，回一个 ACK 把当前窗口带回去
k.acklist = append(k.acklist, ackItem{seg.sn, seg.ts})
}
return 0
}
// Update 由上层按固定节拍调用，驱动定时 flush 与重传。
func (k *KCP) Update(now uint32) {
k.now = now
if k.tsFlush == 0 {
k.tsFlush = now
}
if itimediff(now, k.tsFlush) >= 0 {
k.tsFlush = now + k.interval
k.flush()
}
}
// flush 是协议的心脏：搬数据进窗口、发 ACK、决定重传。
func (k *KCP) flush() {
if k.output == nil {
return
}
limit := k.sndWnd
if k.rmtWnd < limit {
limit = k.rmtWnd
}
for len(k.sndQueue) > 0 && uint32(len(k.sndBuf)) < limit {
seg := k.sndQueue[0]
k.sndQueue = k.sndQueue[1:]
seg.conv, seg.cmd, seg.wnd = k.conv, cmdPush, uint16(k.rcvWnd)
seg.ts, seg.sn, seg.una = k.now, k.sndNxt, k.rcvNxt
seg.rto, seg.resendts, seg.xmit, seg.fastack = uint32(k.rto), k.now+uint32(k.rto), 0, 0
k.sndNxt++
k.sndBuf = append(k.sndBuf, seg)
}
for _, a := range k.acklist {
ack := &segment{conv: k.conv, cmd: cmdAck, wnd: uint16(k.rcvWnd), una: k.rcvNxt, sn: a.sn, ts: a.ts}
k.output(ack.encode())
}
k.acklist = k.acklist[:0]
for _, seg := range k.sndBuf {
needsend := false
switch {
case seg.xmit == 0: // 首次发送
needsend = true
seg.xmit++
seg.rto = uint32(k.rto)
seg.resendts = k.now + seg.rto
case itimediff(k.now, seg.resendts) >= 0: // 超时重传，退避半格
needsend = true
seg.xmit++
seg.rto += uint32(k.rto) / 2
if seg.rto < minRTO {
seg.rto = minRTO
}
seg.resendts = k.now + seg.rto
k.retransmits++
case k.fastresend > 0 && int(seg.fastack) >= k.fastresend: // 快速重传
needsend = true
seg.xmit++
seg.fastack = 0
seg.resendts = k.now + seg.rto
k.retransmits++
}
if needsend {
seg.ts, seg.wnd, seg.una = k.now, uint16(k.rcvWnd), k.rcvNxt
k.output(seg.encode())
}
}
if k.rmtWnd == 0 { // 对端窗口为 0 时主动探测，避免双方互相等死
probe := &segment{conv: k.conv, cmd: cmdWask, wnd: uint16(k.rcvWnd), una: k.rcvNxt, sn: k.sndNxt}
k.output(probe.encode())
}
}
// parseUna 处理累积确认：una 之前的段都已到达。
func (k *KCP) parseUna(una uint32) {
for len(k.sndBuf) > 0 && itimediff(una, k.sndBuf[0].sn) > 0 {
k.sndBuf = k.sndBuf[1:]
}
}
// parseAck 摘掉被确认的段，并给被它跳过的更早的段记一次 fastack。
func (k *KCP) parseAck(sn uint32) {
if itimediff(sn, k.sndUna) < 0 || itimediff(sn, k.sndNxt) >= 0 {
return
}
for i, seg := range k.sndBuf {
if itimediff(sn, seg.sn) < 0 {
return
}
if sn == seg.sn {
k.sndBuf = append(k.sndBuf[:i], k.sndBuf[i+1:]...)
return
}
seg.fastack++
}
}
// parseData 把数据段插入接收缓冲：超窗丢弃、重复去重、按 sn 重排，
// 再把连续的段推进到交付队列。
func (k *KCP) parseData(seg *segment) {
if itimediff(seg.sn, k.rcvNxt) < 0 {
return
}
for _, s := range k.rcvBuf {
if s.sn == seg.sn {
return
}
}
k.rcvBuf = append(k.rcvBuf, seg)
sort.Slice(k.rcvBuf, func(i, j int) bool { return itimediff(k.rcvBuf[i].sn, k.rcvBuf[j].sn) < 0 })
for len(k.rcvBuf) > 0 && k.rcvBuf[0].sn == k.rcvNxt {
k.rcvQueue = append(k.rcvQueue, k.rcvBuf[0])
k.rcvBuf = k.rcvBuf[1:]
k.rcvNxt++
}
}
// shrinkBuf 让 sndUna 始终指向最早未确认的段。
func (k *KCP) shrinkBuf() {
if len(k.sndBuf) > 0 {
k.sndUna = k.sndBuf[0].sn
} else {
k.sndUna = k.sndNxt
}
}
// updateRTT 用 Jacobson 算法更新 SRTT、RTTVAR 与 RTO。
func (k *KCP) updateRTT(rtt int32) {
if k.srtt == 0 {
k.srtt, k.rttvar = rtt, rtt/2
} else {
delta := rtt - k.srtt
k.srtt += delta / 8
if delta < 0 {
delta = -delta
}
k.rttvar += (delta - k.rttvar) / 4
}
rto := k.srtt + 4*k.rttvar
if rto < minRTO {
rto = minRTO
}
if rto > maxRTO {
rto = maxRTO
}
k.rto = rto
}
Send 只入队，Update 才按 interval 节拍 flush：一次 flush 做四件事——把窗口内的段从 sndQueue 搬进 sndBuf 并分配 sn、把累积的 ACK 发出去、遍历 sndBuf 决定谁该重传、窗口为 0 时发探测。Input 则相反：先处理 una 与 ACK，再把数据段按 sn 插进 rcvBuf，只把连续的一段推进 rcvQueue 交付。
ARQ：RTO、滑动窗口与快速重传
RTO 估计用 Jacobson 算法：srtt += (rtt - srtt)/8、rttvar += (|rtt-srtt| - rttvar)/4、rto = srtt + 4×rttvar，并夹在 30ms 与 60000ms 之间。RTT 样本来自 ACK 回显的 ts：发送方打时间戳，接收方原样带回，发送方用 now - ts 得到一次往返。
滑动窗口由三个数决定：sndWnd（本端在途上限）、rmtWnd（对端通告的接收窗口）、rcvWnd（本端接收窗口），flush 里实际允许的在途量取 min(sndWnd, rmtWnd)；una 是累积确认——收到 una = k 就意味着 < k 的段全部到达，可直接从 sndBuf 头部成批摘除，不必逐条 ACK。快速重传则靠 fastack 计数：当某个 sn 的 ACK 到达时，所有序号更小、仍未确认的段 fastack++；一旦 fastack >= fastresend（默认 2），说明对端已收到后面的段、只差这一个，于是立刻重传，不必等 RTO 到期——这是弱网下把 P99 压下去的关键，5.3 的数字会印证。
测试：五个必须覆盖的场景
internal/kcp/kcp_test.go 用一条带延迟与丢包的虚拟链路把两个端点连起来，虚拟时钟每毫秒推进一次，覆盖五个场景：正常收发（TestSendRecvBasic）、50% 丢包下重传最终送达（TestLossRetransmitDelivers）、乱序投递后按 sn 重排（TestOutOfOrderReassembly，把三段按 3-1-2 喂给接收端）、重复包去重（TestDuplicateDedup，同一个包喂三遍只交付一次）、RTO 收敛（TestRTOConvergesToRTT）。
$ go test -v ./internal/kcp/ # 5 个用例全 PASS
--- PASS: TestSendRecvBasic | TestOutOfOrderReassembly | TestDuplicateDedup | TestLossRetransmitDelivers | TestRTOConvergesToRTT
kcp_test.go:96: 300B 消息在 50% 丢包下送达，重传 4 次；kcp_test.go:162: RTT=60ms → RTO 98→82ms
PASS
最后一行值得展开：RTO 从默认的 200ms 收敛到 82ms，而链路真实 RTT 是 60ms，多出来的约 20ms 来自 flush 节拍（10ms）与 ACK 要等下一次 flush 才发出——这说明估计在跟随真实 RTT，而不是停在初值。乱序用例里三段按 3-1-2 到达，Recv 仍依次返回 m1/m2/m3，靠的是 rcvBuf 排序加连续性检查。
小结
这一节把可靠 UDP 拆成了四个零件：报文头（谁是谁）、ARQ（丢了怎么办）、窗口（一次能发多少）、去重与重排（重复和乱序怎么办）。它有意丢掉了拥塞控制与连接协商，所以不能直接上生产，但保留了实时对战最需要的那部分语义，且每个决定都能在测试里被验证。下一节把这个实现放进一个确定性的弱网模拟器，在 0%/1%/5%/20% 四档丢包下测消息延迟 P50/P99 与冗余发送率，并和 5.1 的单流 TCP 式基线正面对比。
阅读导航：上一节：5.1 为什么游戏不只用 TCP · 下一节：5.3 丢包与弱网模拟实测 。
继续阅读
探索更多技术文章
浏览归档，发现更多关于系统设计、工具链和工程实践的内容。