简介MPSP 是一个以 Java 实现的多协议服务处理开源项目面向网络服务开发者、Java 后端工程师及对协议解析感兴趣的进阶学习者源码覆盖 TCP/IP、UDP、HTTP 等常见协议的并发处理场景。资源包共 23 个文件以 15 个 Java 源文件为主体另有 Gradle 构建与属性配置文件、启动脚本和 Markdown 帮助文档整体仅 18KB结构紧凑、目录清晰适合快速通读与二次开发。目前已有 443 人浏览学习该项目。借助源码可系统掌握 Java 网络编程java.net/java.nio、多线程与 ExecutorService、协议解析逻辑以及事件驱动、工厂/单例/观察者等设计模式的实际运用同时还能看到 Gradle 构建配置、单元测试组织、日志与版本忽略规则等工程化细节对从零搭建多协议处理服务或理解高性能网络框架均具有直接的参考价值。1. MPSP 是什么一个多链路并发传输协议而不是又一个 MQTT 外壳MPSP 这个缩写在我手里的意思是 Multi-Path Secure Protocol一套多链路并发传输协议。它来自一次真实的现场事故设备通过 Wi-Fi 上报数据一下雨现场 Wi-Fi 就掉线切到 4G 又开始丢包单链路怎么调都救不回来。于是我们换了个思路——不再“选一条好链路”而是“多链路同时发谁先到用谁”。MPSP 没有标准文档也没后台组织背书它解决的是弱网下最后一公里的可靠性问题。适合物联网网关、边缘视频推流、车机和需要跨多个网络制式传输的场景。如果你正被“信号满格但数据回不来”折磨这套方案值得抄走改改。2. 协议头设计帧结构、序号与校验字段这样定才扛得住弱网任何协议先立头。MPSP 的压力来自多链路同时到达同一个数据帧可能经 Wi-Fi 和 4G 各来一份也可能前一半走 4G、后一半走 Wi-Fi。所以帧头必须让接收方明确知道三件事这是哪个数据帧、从哪条链路来、属于哪个会话。我们在改版三次后把 MPSP 帧定成了 16 字节固定头 4 字节 CRC 负载。2.1 16 字节固定头Type、Seq、Ack、PathID 这样排字段设计如下表字段字节数说明magic2固定 0x5AA5用于过滤噪声包type1帧类型SYN / SYN-ACK / ACK / DATA / NACK / PATH_PROBEflags1位标记bit0会话建立bit1结束bit2还有后续分片path_id1链路 ID1Wi-Fi2蜂窝3以太网res1预留握手后存放 session_id 的低位seq4全局发送序号小端ack4期望收到的下一个序号小端payload_len2CRC负载总长度单位字节seq 和 ack 都用 32 位无符号不是随意拍脑袋。现场缓冲可能开到几百 KB16 位 seq 在连续高速传输下很快就会回绕导致新帧和旧帧重号。32 位在正常会话时长内基本不会回绕回绕也能通过 flag 位处理。path_id 只需要 1 字节因为真实链路数不超过十几条更大值保留给内部隧道扩展。magic 的作用经常被低估。现场总线上并不只跑 MPSP还有设备的私有调试协议和广播报文。没有 magic接收端会把别人家的 0x01 当成 SYN然后白等 8 秒。有了 magic解析函数可以在读前两个字节后直接丢包节省大量 CPU。2.2 为什么不用 MD5CRC32 会话密钥的校验收敛有人会问帧校验为什么不用 MD5 甚至 SHA因为 MPSP 的目标场景是低功耗 MCU 和高吞吐边缘网关。MD5 摘要 16 字节比整个头部还长SHA 更贵。弱网下的误码是随机比特翻转不是密码学对抗。CRC32 检测随机误码的能力足够而真正的机密性交给会话加密。我们的做法是握手完成后两端用 token 派生 AES-CTR 会话密钥。每个分片先加密再对密文算 CRC32。这样一帧数据从设备到服务端任何一比特被改CRC 直接失败有人截包也解不开内容。AES-CTR 的 nonce 由 magic session_id seq 组成保证同一会话内每帧 nonce 不同。这里不要把 nonce 设计成纯 seq 偏移否则两个场景交错时容易撞 nonce。如果项目暂时不需要加密可以先把 AES-CTR 那层关掉只用 CRC32 校验收发。但我会建议保留密钥派生的框架而不是彻底删掉。因为现场一旦要求加解密你不需要改帧格式只需要在 build_frame 前后各加一层变换。这个习惯会让 MPSP 的演进省很多事。2.3 最小协议实现构造/解析 MPSP 帧的 Python 代码import struct, zlib MAGIC 0x5AA5 TYPE_SYN 1 TYPE_DATA 4 def build_frame(type_, seq, ack, path_id, payloadb): crc zlib.crc32(payload) 0xffffffff body struct.pack(I, crc) payload length len(body) header struct.pack(HBBBBIIH, MAGIC, type_, 0, path_id, 0, seq, ack, length) return header body def parse_frame(buf): if len(buf) 16: raise ValueError(short frame) magic, type_, flags, path_id, res, seq, ack, length struct.unpack(HBBBBIIH, buf[:16]) if magic ! MAGIC: raise ValueError(bad magic) if len(buf) 16 length: raise ValueError(incomplete payload) crc_stored struct.unpack(I, buf[16:20])[0] payload buf[20:16length] if zlib.crc32(payload) 0xffffffff ! crc_stored: raise ValueError(crc mismatch) return type_, flags, path_id, seq, ack, payload逻辑说明build_frame 先算 payload 的 CRC32再把 CRC 和 payload 合成 body最后拼 16 字节头。parse_frame 第一步检查 16 字节固定头第二步检查完整长度第三步才校验 CRC。这里故意把 CRC 检查放在长度检查之后避免对不完整缓冲区做多余计算。参数说明MAGIC 固定 0x5AA5大小端互换后是 0xA55A如果对端收到串了字节顺序会立刻暴露。TYPE_SYN 和 TYPE_DATA 只是最小示例实际实现至少有 6 种 type完整枚举见第 3 章。res 字段在第 2.2 节的加密方案里充当 session_id 的低位。如果会话数超过 255可以用整个 res 作为会话表索引但那样就失去了设备 ID 的熵建议扩展 header。注意 payload_len 包含 4 字节 CRC。很多初写协议的人喜欢让 len 只表示 payload导致接收端解析 CRC 时多取或少取 4 字节这是常见坑。如果这段代码放到真实工程里还要处理一个底层细节当使用 TCP 或者 Unix socket 承载 MPSP 帧时recv 可能只读到半个头。parse_frame 里直接抛异常并不够应该在读取循环里先把字节攒够再解析。第 5 章会讲这种“半包”怎么处理。3. 握手与路径探测会话怎么建立多链路怎么确认可用MPSP 虽然叫协议但它并不像 TCP 那样每个连接只绑一个四元组。设备侧同时有 Wi-Fi、4G、以太网三个网络接口每个接口都可能向服务端建连所以 MPSP 会话是“一个逻辑会话承载在多个物理链路之上”。这就需要在握手阶段把每条链路的地址信息都告诉服务端并让服务端知道每条链路的实时质量。3.1 握手流程Syn、Syn-Ack、Ack 加 PathProbe握手分成四个阶段设备从主链路例如 Wi-Fi发 SYNpayload 里带设备 ID、固件版本、可用链路列表。服务端查重后返回 SYN-ACK分配 session_id并下发 16 字节随机 token。设备收到 SYN-ACK从所有可用链路各发一个 ACK PathProbe。PathProbe 的 path_id 就是链路本身payload 带 probe_id 和当前毫秒时间戳。服务端收到 PathProbe记录该链路的源地址和 RTT 初值会话进入 EST 状态。这里最关键的是第 3 步。如果 SYN-ACK 只从 Wi-Fi 回4G 链路没有收到过任何服务端包它就不可能被当作有效路径。让设备从所有链路主动发 probe服务端才知道哪些路径真正通了。不要在服务端“猜”哪条链路可用必须用探针实测。3.2 路径探测与权重分配RTT 和丢包率怎么换算MPSP 每 5 秒发一次 probe统计最近 10 次 RTT 样本。权重计算采用一个虚拟毫秒模型weight[path] 1 / (avg_rtt 5 * jitter 800 * loss)其中 avg_rtt 是加权平均 RTT单位毫秒jitter 是样本标准差loss 是最近 10 次 probe 的丢包比例。分母越大路径越差权重越低。这个公式不是拍脑袋它把一次丢包等价于 800ms 额外延迟避免“延迟低但丢包率 20%”的路径被误选为最优。import math def calc_weight(samples): total len(samples) if total 0: return 0.0 valid [s for s in samples if s is not None] if not valid: return 0.0 avg_rtt sum(valid) / len(valid) variance sum((s - avg_rtt) ** 2 for s in valid) / len(valid) jitter math.sqrt(variance) loss (total - len(valid)) / total return 1.0 / (avg_rtt 5.0 * jitter 800.0 * loss 1e-6)逻辑说明samples 里 None 表示超时的 probevalid 只取实际 RTT。loss 按 None 在总数里的比例计算。1e-6 是防零保护保证所有样本都有效且延迟极低时不会除零。返回的权重是浮点数发送端按权重比例做分片分配但最少给每条链路 10% 的发送量这是踩坑后的保护。实际部署时如果发现某条链路权重长期为 0不要立刻移除它。在工业现场最差的链路往往是关键时刻的救命通道。我们会在发送策略里规定每条链路每 10 个分片至少分到 1 个保留它的探针作用。3.3 服务端会话表维护过期、重放、并发上限服务端会话表用字典实现即可关键是不能只按源 IP 识别会话。设备从 Wi-Fi 切到 4G 后源 IP 会变按 IP 找会话就是必现的 bug。正确做法是把设备 ID 作为主键源地址只作为路径表里的一个字段。import time SESSION_TIMEOUT 30 MAX_SESSIONS 1024 sessions {} def get_or_create_session(device_id): if device_id not in sessions: if len(sessions) MAX_SESSIONS: old_key min(sessions, keylambda k: sessions[k][last_active]) del sessions[old_key] sessions[device_id] { state: SYN_RCVD, paths: {}, key: b, last_active: time.time(), next_seq: 0 } return sessions[device_id] def cleanup_sessions(nowNone): now now or time.time() for dev_id in list(sessions): if now - sessions[dev_id][last_active] SESSION_TIMEOUT: del sessions[dev_id]逻辑说明get_or_create_session 在并发数超过 MAX_SESSIONS 时用 min 找出 last_active 最小的会话删除。这里假设字典里只有几千个会话线性扫描可以接受如果上十万级就要换成小顶堆或 LRU 链。参数说明SESSION_TIMEOUT 设 30 秒比现场网络最长断线时间 15 秒长又不会让僵尸会话占太久。MAX_SESSIONS 1024 是单机经验值服务端每会话会维护 key 和路径表开太大大概率在内存上出问题。state 字段可以从 SYN_RCVD 到 EST最后到 FIN实际使用时建议做成状态机避免收到 FIN 后还能收数据。关于重放防护如果同一个 device_id 发来重复 SYN而旧会话仍活跃不要直接覆盖。我们做法是让新连接先进入 SYN_WAIT等待 5 秒。如果新连接能连续发两次有效 ACK说明设备端确实重启了再切换会话。这能挡掉抓包重放也能避免 Wi-Fi 和蜂窝两个模块同时发 SYN 时服务端左右横跳。4. 滑动窗口与重传把多链路的乱序包拼回有序数据MPSP 的 seq 是全局统一的也就是说无论这份数据从 Wi-Fi 来还是从 4G 来它只有一个编号。这样接收端不需要做每链路缓冲只需一个全局滑窗。代价是发送端必须清楚每一个 seq 发到了哪条链路避免同一帧重复写入不同 path 造成带宽浪费。这一章给出一套可运行的接收/发送缓冲逻辑。4.1 接收缓冲与乱序恢复Seq 滑窗的边界条件接收端维护 expected_seq 和 window_size。收到一个 seq如果落在 [expected, expected window_size) 就缓存等于 expected 就连续交付小于 expected 但是旧帧回一个 ACK 告诉对端当前游标即可大于 expected window_size 说明发送端滑窗推得太快也回 ACK不要乱丢。class MpspReceiver: def __init__(self, window_size256): self.window_size window_size self.expected 0 self.buf {} def on_data(self, seq, ack, payload): if seq self.expected: return old_ack, self.expected if seq self.expected self.window_size: return out_of_window, self.expected self.buf[seq] (ack, payload) while self.expected in self.buf: ack, payload self.buf.pop(self.expected) yield ack, payload self.expected 1 return buffered, self.expected逻辑说明yield 是关键设计。它把连续可交付的数据一条条抛给上层而不是先拼成一个 list 再返回。这样内存占用只跟乱序缺口有关不会因为窗口大而一次性暴涨。返回的字符串只是状态码用于打日志不要把它当协议状态。参数说明window_size 256 在每帧 1KB 的情况下缓冲区占用约 256KB。MCU 端建议降到 128服务端可以到 512。seq 回绕问题MPSP 约定同一会话内窗口跨度不能超过 131所以把回绕留在内存里做无符号比较即可。Python 整数没有溢出但要小心在 C/C 端用有符号 int 比较那会翻车。4.2 重传策略选择性 NACK 还是定时重传在多链路弱网下我们最后采用的是“定时重传 NACK 共存”而不是只选一种。纯定时重传简单但链路抖动大时恢复慢纯 NACK 依赖反向链路反向链路本身就是弱链路时会雪上加霜。NACK 的触发时机接收端发现某个 seq 缺失立即回一个 NACK 帧payload 是 4 字节的缺失 seq。发送端收到 NACK 后如果该 seq 仍在自己 unacked 里就把 send_time 置 0让它在下一个 tick 被重传。这样 NACK 只是提前触发不替代定时器。class MpspSender: def __init__(self, window_size256, rto0.8): self.window_size window_size self.rto rto self.unacked {} def on_ack(self, seq): if seq in self.unacked: del self.unacked[seq] def on_nack(self, missing_seq): if missing_seq in self.unacked: self.unacked[missing_seq] (0, self.unacked[missing_seq][1]) def tick(self, now, send_cb): for seq, (t0, payload) in list(self.unacked.items()): if now - t0 self.rto: self.unacked[seq] (now, payload) send_cb(seq, payload)逻辑说明on_ack 直接把已确认的帧删掉unacked 里剩下的就是超时未确认。on_nack 不额外建队列因为一旦把 send_time 改为 0下一个 tick 必然触发重传比单独处理 NACK 更省代码。send_cb 由外部传入把重传帧交给哪条路径由发送策略决定。参数说明rto 0.8 秒是针对 4G 链路平均 RTT 40ms 场景。如果主要跑卫星链路RTT 500msrto 至少设 1.5 秒。太短会导致所有链路都忙于重传吞吐断崖。tick 建议 200ms 跑一次不要让调用方自己决定间隔。否则高延迟路径上的帧要等到下一个 200ms 才被检查反而放大抖动。4.3 代码复现一个可跑的发送端与接收端框架一个最小发送端循环如下。这里为演示把一份文件的所有分片轮流分配给一个 path 发送同时记录每个 seq 到 unacked。def send_frames(sock, peer, payloads, path_ids): sender MpspSender() seq 0 for payload in payloads: path_id path_ids[seq % len(path_ids)] sock.sendto(build_frame(TYPE_DATA, seq, sender.expected, path_id, payload), peer) sender.unacked[seq] (time.time(), payload) seq 1 while sender.unacked: sender.tick(time.time(), lambda s, p: sock.sendto(build_frame(TYPE_DATA, s, sender.expected, s % len(path_ids), p), peer)) time.sleep(0.01)逻辑说明发送端把 payloads 里每个分片按 seq 轮询分给不同 path接收端用 4.1 的 MpspReceiver 把分散的包按序重组。这里的 sleep 0.01 是为了防止 tick 在 CPU 上跑成死循环真实工程里可以用 select 或 event loop。参数说明分片大小建议 1400 字节。比如你在 UDP 上跑 MPSPMTU 1500 减去 IP 头 20 和 UDP 头 81400 是安全边界。设大了会触发 IP 分片反而引入额外乱序。冗余模式怎么做把上面循环里的 sendto 改成对每个 path_id 都发一次流量变成 N 倍但丢包率大幅下降。建议控制命令走冗余模式大块文件走轮询分片模式二者切换由应用层决定。对接收端来说只需要持续调用 parse_frame然后把 type 为 DATA 的帧交给 MpspReceiver 即可。乱序缺帧会阻塞 expected 前进直到超时重传补上。这也是第 5 章要讲的一个关键排查点。5. MPSP 弱网避坑5 个我踩过的稳定性问题协议能跑通只用了半天调稳定用了两天。这章的坑每一个都让现场翻过车按现象、原因、解决记录希望能给你当排查手册。5.1 链路切换瞬间握手失效怎么办现象设备从 Wi-Fi 切到 4G 后服务端一直等旧地址的 Ack新地址的帧全被当成非法连接丢弃直到会话超时重建。原因服务端把“源 IP:端口”当成了会话唯一标识。设备一切网地址变了会话就找不到了。解决把会话主键改成设备 ID源地址只作为路径表中的一条可达地址。MPSP 握手时用设备 ID 注册之后每次帧头里都带 session_id我们用 res 字段放低位。链路切换后新地址的帧到达时服务端能通过 session_id 找到旧会话并自动把新地址加入该会话的路径表。这样切换不会重建会话只是多了一条 path。5.2 粘包与半包取码器必须处理短读现象用 TCP 承载 MPSP 帧时接收端按 164len 直接 recv结果一次 recv 返回 20 字节或者返回 0程序卡死。原因recv 不保证一次返回完整帧尤其是帧头被拆开时。如果代码只调用一次 recv读到半个头就解析必然出错。解决写一个 read_exact 函数先读 16 字节头解析出 payload_len再循环读到 buffer 满。代码量不大但很多人都跳过这一步直接信任 recv。遇到短帧时不要立刻报错先把没读完的字节攒到下一轮这就是“半包”处理。5.3 时钟不同步导致的时间戳与续传冲突现象设备端用本地时钟写入时间戳服务端发现设备时间比服务器快 5 分钟把一部分合法帧当成未来帧丢弃。原因MPSP 早期版本用时间戳做去重跨设备时钟不准时误判。解决去重只依赖 seq时间戳只用于 RTT 统计和诊断。需要续传时用 ack seq 作为游标不要用时间戳拼接。领导看到日志里时间戳乱不要慌说明设备 RTC 没同步和协议无关。5.4 多路径并发导致的对端重复消费现象一条数据同时从 Wi-Fi 和 4G 发送接收端应用层发现同一时序数据写入了两条记录。原因冗余模式发送时同一个 seq 在不同路径各收到一次接收端没有做 seq 去重直接把每帧都交给上层。解决把第 4 章的滑窗去重放到应用层最前面——只有被 expected 弹出的帧才交给上层其它重复帧即使校验通过也直接丢弃。应用层要尽量幂等不要用“插入时间”做唯一键。我的一个同事为了让字段好看用 arrival_time 做唯一索引结果压力测试下 duplicate key 直接崩了。5.5 路由策略与协议无关别让系统自动路由拆散绑定现象服务器有两个网卡设备 Wi-Fi 和 4G 的流量到达后服务端回包走默认路由结果两个链路都收到同一个路径的响应负载测试数据很难看。原因回包路径没有绑定到接收入口内核自动路由把响应都从默认网关发出。解决服务端每个会话的路径表中记录“该链路最先到达时的本地接收接口”回包时用不同的 socket 或设置 IP_PKTINFO / SO_BINDTODEVICE 指定发送接口。如果不想做这么底层至少在应用层记录 path_id 与对方地址的映射别让内核的默认路由破坏多路径语义。这个坑很容易被误判为“协议问题”其实协议本身没毛病。6. 压测 MPSP 的几种现实手段从丢包注入到会话恢复协议写完之后必须压测不能只看演示通不通。我通常在 Linux 上用 tc netem 注入丢包和延迟再在 Python 端跑一个整合脚本统计吞吐、重传率和 out_of_window 次数。下面是常用的两组命令。# 模拟 Wi-Fi 链路5% 丢包、40ms 延迟、10ms jitter sudo tc qdisc add dev wlan0 root netem loss 5% delay 40ms 10ms distribution normal # 模拟蜂窝链路2% 丢包、80ms 延迟 sudo tc qdisc add dev usb0 root netem loss 2% delay 80ms逻辑说明netem 是 Linux 自带的网络模拟器不需要额外工具。压测时把服务端放到另一台机器客户端通过两个网络接口同时连接持续发 5 分钟数据统计最终交付的字节数 / 原始发送字节数得到有效吞吐。观察重传率如果 rto 调太短会发现发的数据里有一大半是重传包如果窗口调太大弱网下乱序数量会暴涨。我还习惯在压测中途拔掉其中一根网线观察会话是否在下一轮路径探测后自动恢复。设备端路径探测周期 5 秒路径表删除条件设为“连续 3 次 probe 失败”拔线后最长 15 秒内会从剩余链路把积压数据补完。这个恢复时间可以用下面的脚本验证# 每 5 秒打印一次接收端 expected_seq拔线后如果 4 个周期内还停在原地就有问题 while True: time.sleep(5) print(expected, receiver.expected, buffered, len(receiver.buf))说明如果 buffered 一直增长而 expected 不动说明有链路在接收但数据传不到应用层优先检查乱序窗口是否被某个缺失 seq 堵住。这就是我碰到过最典型的“看似连着实际死了”的假活状态。事后我把 NACK 改成对阻塞 seq 每 100ms 发一次而不是等发送端超时问题就解决了。以上就是 MPSP 从我手上落地到现场的全过程。协议不复杂但弱网下的细节比功能代码更难。希望帮到你。本文还有配套的精品资源点击获取