4. 基于UDP的冗余传输

任务目标

在SOME/IP通信中，UDP天然不保证消息可靠送达。本案例通过发送端对同一SOME/IP消息进行多次冗余发送，接收端基于消息ID与序列号进行去重，从而在应用层提升消息到达率。核心目标是实现一个最小可运行的冗余发送与去重接收闭环。

实现方案

发送端维护一个递增的序列号，将同一序列号的消息连续发送3次（冗余因子=3）。接收端维护一个最近处理过的序列号集合，仅当收到更大（更新）的序列号时才向上层交付消息，重复序列号直接丢弃。为简化演示，SOME/IP消息体仅包含序列号字段，实际工程中可替换为完整SOME/IP头部。

完整代码（Python）

import socket
import threading
import time
from collections import deque

# SOME/IP消息格式：4字节消息ID + 4字节序列号（简化）
MSG_ID = 0x1234
REDUNDANCY = 3
LOCAL_IP = "127.0.0.1"
PORT = 5001

def build_message(seq):
    """构造简化SOME/IP消息：消息ID(4字节) + 序列号(4字节)"""
    return MSG_ID.to_bytes(4, 'big') + seq.to_bytes(4, 'big')

def parse_message(data):
    """解析消息，返回(消息ID, 序列号)"""
    msg_id = int.from_bytes(data[:4], 'big')
    seq = int.from_bytes(data[4:8], 'big')
    return msg_id, seq

def sender():
    """发送端：每个序列号连续发送REDUNDANCY次"""
    sock = socket.socket(socket.AF_INET, socket.AF_INET)
    sock.connect((LOCAL_IP, PORT))
    for seq in range(1, 6):  # 发送5个不同消息
        payload = build_message(seq)
        for _ in range(REDUNDANCY):
            sock.send(payload)
            time.sleep(0.01)  # 模拟网络间隔
        time.sleep(0.05)  # 消息间间隔
    sock.close()

def receiver():
    """接收端：去重后仅交付新序列号消息"""
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.bind((LOCAL_IP, PORT))
    sock.settimeout(2.0)
    
    # 用双端队列保存最近已处理的序列号（窗口大小=10）
    recent_seqs = deque(maxlen=10)
    delivered = []
    
    end_time = time.time() + 2.5
    while time.time() < end_time:
        try:
            data, _ = sock.recvfrom(1024)
            msg_id, seq = parse_message(data)
            if msg_id != MSG_ID:
                continue
            if seq not in recent_seqs:
                recent_seqs.append(seq)
                delivered.append(seq)
                print(f"[交付] 序列号 {seq}")
            else:
                print(f"[丢弃] 重复序列号 {seq}")
        except socket.timeout:
            break
    sock.close()
    return delivered

if __name__ == "__main__":
    # 启动接收线程
    recv_thread = threading.Thread(target=receiver)
    recv_thread.start()
    time.sleep(0.1)  # 确保接收端就绪
    sender()
    recv_thread.join()
    print("演示完成")

运行方式

  1. 保存代码为 redundant_udp.py，使用Python 3.6+运行：python redundant_udp.py
  2. 程序自动启动接收线程，随后发送端发送5个不同序列号的消息，每个消息重复3次

验证方法与通过标准

  • 预期输出特征：控制台应打印5行[交付]（序列号1~5各一次），以及10行[丢弃]（每个序列号重复2次被丢弃）
  • 判定通过：交付序列号恰好为1,2,3,4,5且无重复；丢弃消息均为已交付序列号的重复副本
  • 可靠性验证：若在发送端注释掉冗余发送（仅发送1次），接收端仍能收到全部5条消息（因本地回环不丢包）；如需模拟丢包，可在发送循环中随机跳过某次发送，观察冗余机制是否保证消息仍被交付

关键说明

  • 去重窗口（maxlen=10）用于防止序列号无限增长，实际工程中应根据消息频率和乱序范围调整
  • 本实现使用UDP回环地址验证逻辑正确性；真实网络环境中，冗余发送可显著降低丢包导致的业务中断概率
  • 序列号溢出处理（超过2^32）未在演示中覆盖，生产实现需考虑回绕场景
← 上一章 📖 返回目录 下一章 →