4. SOME/IP服务发现重传机制

任务目标

在SOME/IP服务发现（SD）中，服务提供者通过发送OfferService报文宣告服务可用，服务消费者通过发送FindService报文搜索服务。由于UDP传输不可靠，报文可能丢失，因此需要定时重传机制确保服务信息可靠传播。本案例实现一个最小可复现的SD重传管理器，支持OfferService与FindService报文的定时重传，并验证重传次数与间隔符合预期。

核心实现

以下代码实现一个轻量级重传管理器，核心逻辑包括：报文发送记录（跟踪已发送但未确认的报文）、定时重传调度（按初始间隔与最大重传次数执行）、确认取消机制（收到匹配响应后停止重传）。为聚焦重传逻辑本身，不引入完整SOME/IP序列化，使用简化报文结构体表示关键字段。

import time
import threading
from dataclasses import dataclass
from enum import Enum
from typing import Optional, Callable, Dict

class MessageType(Enum):
    OFFER_SERVICE = 0x01
    FIND_SERVICE = 0x02

@dataclass
class SDMessage:
    msg_type: MessageType
    service_id: int
    instance_id: int
    session_id: int

class RetransmissionManager:
    """SOME/IP SD 报文定时重传管理器"""
    
    def __init__(self, initial_delay: float = 1.0, max_retries: int = 3, 
                 send_callback: Optional[Callable[[SDMessage], None]] = None):
        self.initial_delay = initial_delay      # 初始重传间隔（秒）
        self.max_retries = max_retries          # 最大重传次数
        self.send_callback = send_callback      # 实际发送报文的回调
        self._records: Dict[int, dict] = {}     # session_id -> 记录
        self._lock = threading.Lock()
        self._stop_event = threading.Event()
        self._threads: list[threading.Thread] = []
    
    def send_message(self, msg: SDMessage) -> None:
        """首次发送报文并启动重传定时器"""
        with self._lock:
            self._records[msg.session_id] = {
                'msg': msg,
                'retries_left': self.max_retries,
                'delay': self.initial_delay
            }
        # 立即发送首次报文
        if self.send_callback:
            self.send_callback(msg)
        # 启动重传线程
        t = threading.Thread(target=self._retransmit_loop, args=(msg.session_id,), daemon=True)
        self._threads.append(t)
        t.start()
    
    def confirm_received(self, session_id: int) -> None:
        """收到对应响应（如OfferService确认），取消重传"""
        with self._lock:
            self._records.pop(session_id, None)
    
    def _retransmit_loop(self, session_id: int) -> None:
        """重传循环：按间隔重发，直到确认或达到最大次数"""
        while not self._stop_event.is_set():
            with self._lock:
                record = self._records.get(session_id)
                if record is None:
                    return  # 已确认，停止
                if record['retries_left'] <= 0:
                    self._records.pop(session_id, None)
                    return  # 重传次数耗尽
                record['retries_left'] -= 1
                delay = record['delay']
                msg = record['msg']
            
            # 等待间隔时间（可被停止事件中断）
            if self._stop_event.wait(delay):
                return
            
            # 执行重传
            if self.send_callback:
                self.send_callback(msg)
    
    def stop(self) -> None:
        """停止所有重传线程"""
        self._stop_event.set()
        for t in self._threads:
            t.join(timeout=0.5)
        self._threads.clear()

# ========== 验证示例 ==========
if __name__ == '__main__':
    sent_log = []
    
    def mock_send(msg: SDMessage):
        sent_log.append((time.time(), msg))
        print(f"[发送] {msg.msg_type.name} service={msg.service_id} "
              f"session={msg.session_id} retries_left={msg.session_id}")
    
    # 场景1：OfferService 重传3次后放弃（无确认）
    mgr = RetransmissionManager(initial_delay=0.1, max_retries=3, send_callback=mock_send)
    offer = SDMessage(MessageType.OFFER_SERVICE, service_id=0x1234, instance_id=0x01, session_id=1)
    mgr.send_message(offer)
    time.sleep(0.5)  # 等待重传完成
    mgr.stop()
    
    offer_count = sum(1 for _, m in sent_log if m.msg_type == MessageType.OFFER_SERVICE)
    print(f"\nOfferService 实际发送次数: {offer_count} (预期: 4 = 首次 + 3次重传)")
    
    # 场景2：FindService 收到确认后停止重传
    sent_log.clear()
    mgr2 = RetransmissionManager(initial_delay=0.1, max_retries=5, send_callback=mock_send)
    find = SDMessage(MessageType.FIND_SERVICE, service_id=0x5678, instance_id=0x02, session_id=2)
    mgr2.send_message(find)
    time.sleep(0.15)  # 等待首次重传前
    mgr2.confirm_received(session_id=2)  # 模拟收到响应
    time.sleep(0.3)
    mgr2.stop()
    
    find_count = sum(1 for _, m in sent_log if m.msg_type == MessageType.FIND_SERVICE)
    print(f"FindService 实际发送次数: {find_count} (预期: 1，确认后不再重传)")

运行与验证

运行方式：直接执行 python sd_retransmission.py（需Python 3.7+，无第三方依赖）。

验证方法：观察输出中的发送日志与统计行：

• 场景1：OfferService应打印4次发送（首次+3次重传），统计行显示实际发送次数: 4，证明无确认时按最大重传次数执行后自动停止。
• 场景2：FindService应仅打印1次发送，统计行显示实际发送次数: 1，证明收到确认后重传被正确取消。

若两个场景统计值分别等于4和1，则重传机制验证通过。可通过修改initial_delay与max_retries参数观察重传间隔与次数变化。

关键设计说明

• 重传策略：采用固定间隔重传（非指数退避），简化实现；实际SOME/IP协议中可扩展为按倍数递增间隔。
• 线程安全：使用threading.Lock保护记录字典，避免并发访问冲突。
• 停止机制：stop()通过事件通知所有重传线程退出，避免程序无法结束。
• 确认匹配：本示例以session_id作为确认匹配键，实际协议中应结合service_id与instance_id。