1. SOME/IP服务发现冗余基础

任务目标

在SOME/IP通信中，服务发现（Service Discovery, SD）负责让客户端动态发现服务实例的位置。本案例实现一个最小化的SD冗余机制：同一服务在多个ECU（节点）上注册，客户端通过SD报文获取全部可用实例，并在首选实例不可达时自动切换至备用实例。

环境与依赖

• Python 3.8+，无第三方库依赖
• 使用UDP Socket模拟SOME/IP SD报文（简化格式，仅保留关键字段）

核心实现

以下代码包含三个角色：服务端节点（两个ECU各自注册同一服务）、SD服务器（维护服务实例表并响应查找）、客户端（发现实例并执行故障切换）。SD报文格式定义为：type|service_id|instance_id|ip|port，其中type为OFFER（服务注册）或FIND（查找服务）。

import socket
import threading
import time
from dataclasses import dataclass
from typing import List, Optional

# ---------- 数据结构 ----------
@dataclass
class ServiceInstance:
    service_id: int
    instance_id: int
    ip: str
    port: int
    alive: bool = True

# ---------- SD服务器（运行于中央节点） ----------
class SDServer:
    def __init__(self, sd_ip: str = "127.0.0.1", sd_port: int = 30490):
        self.sd_ip = sd_ip
        self.sd_port = sd_port
        self.instances: List[ServiceInstance] = []
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.sock.bind((sd_ip, sd_port))
        self.running = True

    def _parse_msg(self, data: bytes) -> List[str]:
        return data.decode().split("|")

    def _handle_offer(self, parts: List[str]):
        # OFFER|service_id|instance_id|ip|port
        _, sid, iid, ip, port = parts
        inst = ServiceInstance(int(sid), int(iid), ip, int(port))
        # 去重：同instance_id只保留最新注册
        self.instances = [i for i in self.instances if i.instance_id != inst.instance_id]
        self.instances.append(inst)
        print(f"[SD] 注册实例: service={sid} instance={iid} @ {ip}:{port}")

    def _handle_find(self, parts: List[str], client_addr):
        # FIND|service_id
        _, sid = parts
        sid = int(sid)
        matches = [i for i in self.instances if i.service_id == sid and i.alive]
        if not matches:
            self.sock.sendto(b"NO_AVAILABLE", client_addr)
            return
        # 返回所有可用实例，格式: FOUND|count|ip:port|ip:port|...
        resp = f"FOUND|{len(matches)}|" + "|".join(f"{i.ip}:{i.port}" for i in matches)
        self.sock.sendto(resp.encode(), client_addr)
        print(f"[SD] 响应查找 service={sid} -> {len(matches)} 个实例")

    def run(self):
        print(f"[SD] 服务发现服务器启动 @ {self.sd_ip}:{self.sd_port}")
        while self.running:
            data, addr = self.sock.recvfrom(1024)
            parts = self._parse_msg(data)
            if parts[0] == "OFFER":
                self._handle_offer(parts)
            elif parts[0] == "FIND":
                self._handle_find(parts, addr)

    def stop(self):
        self.running = False
        self.sock.close()

# ---------- 服务端节点（ECU） ----------
class ServiceNode:
    def __init__(self, node_name: str, service_id: int, instance_id: int,
                 ip: str, port: int, sd_addr: tuple):
        self.node_name = node_name
        self.service_id = service_id
        self.instance_id = instance_id
        self.ip = ip
        self.port = port
        self.sd_addr = sd_addr
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)

    def register(self):
        msg = f"OFFER|{self.service_id}|{self.instance_id}|{self.ip}|{self.port}"
        self.sock.sendto(msg.encode(), self.sd_addr)
        print(f"[{self.node_name}] 已向SD注册 service={self.service_id} instance={self.instance_id}")

    def stop(self):
        self.sock.close()

# ---------- 客户端（带冗余切换） ----------
class SDClient:
    def __init__(self, service_id: int, sd_addr: tuple):
        self.service_id = service_id
        self.sd_addr = sd_addr
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.sock.settimeout(2.0)
        self.candidates: List[tuple] = []  # [(ip, port), ...]
        self.current_idx = 0

    def discover(self) -> bool:
        """向SD发送FIND，获取可用实例列表"""
        msg = f"FIND|{self.service_id}"
        self.sock.sendto(msg.encode(), self.sd_addr)
        try:
            resp, _ = self.sock.recvfrom(1024)
        except socket.timeout:
            print("[Client] 查找超时，无可用服务")
            return False
        parts = resp.decode().split("|")
        if parts[0] == "NO_AVAILABLE":
            print("[Client] SD返回无可用实例")
            return False
        # FOUND|count|ip:port|...
        count = int(parts[1])
        self.candidates = []
        for item in parts[2:2 + count]:
            ip, port = item.split(":")
            self.candidates.append((ip, int(port)))
        self.current_idx = 0
        print(f"[Client] 发现 {count} 个实例: {self.candidates}")
        return True

    def call_service(self) -> bool:
        """尝试调用当前实例，失败则切换到下一个"""
        if not self.candidates:
            return False
        attempts = len(self.candidates)
        for _ in range(attempts):
            ip, port = self.candidates[self.current_idx]
            if self._try_call(ip, port):
                return True
            # 切换至下一实例
            self.current_idx = (self.current_idx + 1) % len(self.candidates)
            print(f"[Client] 切换到备用实例 {self.candidates[self.current_idx]}")
        print("[Client] 所有实例均不可达")
        return False

    def _try_call(self, ip: str, port: int) -> bool:
        """模拟一次服务调用：向服务端口发送PING，收到PONG视为成功"""
        try:
            self.sock.sendto(b"PING", (ip, port))
            self.sock.settimeout(1.0)
            data, _ = self.sock.recvfrom(1024)
            return data == b"PONG"
        except socket.timeout:
            return False
        except OSError:
            return False

    def stop(self):
        self.sock.close()

# ---------- 模拟服务端口（用于验证连通性） ----------
class MockServicePort:
    def __init__(self, ip: str, port: int):
        self.ip = ip
        self.port = port
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.sock.bind((ip, port))
        self.running = True
        threading.Thread(target=self._listen, daemon=True).start()

    def _listen(self):
        while self.running:
            try:
                data, addr = self.sock.recvfrom(1024)
                if data == b"PING":
                    self.sock.sendto(b"PONG", addr)
            except OSError:
                break

    def stop(self):
        self.running = False
        self.sock.close()

# ---------- 主流程（演示冗余切换） ----------
def main():
    SD_ADDR = ("127.0.0.1", 30490)
    SERVICE_ID = 0x1234

    # 启动SD服务器
    sd = SDServer(*SD_ADDR)
    threading.Thread(target=sd.run, daemon=True).start()
    time.sleep(0.2)

    # 两个ECU节点注册同一服务（不同instance_id）
    ecu1 = ServiceNode("ECU1", SERVICE_ID, 1, "127.0.0.1", 5001, SD_ADDR)
    ecu2 = ServiceNode("ECU2", SERVICE_ID, 2, "127.0.0.1", 5002, SD_ADDR)
    ecu1.register()
    ecu2.register()
    time.sleep(0.2)

    # 启动两个模拟服务端口（对应两个实例）
    port1 = MockServicePort("127.0.0.1", 5001)
    port2 = MockServicePort("127.0.0.1", 5002)

    # 客户端发现并调用
    client = SDClient(SERVICE_ID, SD_ADDR)
    if client.discover():
        # 先停止ECU1的模拟服务端口，模拟ECU1故障
        print("\n--- 模拟ECU1故障（关闭5001端口）---")
        port1.stop()
        time.sleep(0.1)

        success = client.call_service()
        print(f"\n调用结果: {'成功' if success else '失败'}")
        if success:
            print(f"当前使用实例: {client.candidates[client.current_idx]}")

    # 清理
    client.stop()
    port2.stop()
    ecu1.stop()
    ecu2.stop()
    sd.stop()

if __name__ == "__main__":
    main()

运行与验证

1. 将上述代码保存为 sd_redundancy.py，直接运行：python3 sd_redundancy.py
2. 观察输出，预期执行流程为：

[SD] 服务发现服务器启动 @ 127.0.0.1:30490
[ECU1] 已向SD注册 service=4660 instance=1
[ECU2] 已向SD注册 service=4660 instance=2
[SD] 注册实例: service=4660 instance=1 @ 127.0.0.1:5001
[SD] 注册实例: service=4660 instance=2 @ 127.0.0.1:5002
[Client] 发现 2 个实例: [('127.0.0.1', 5001), ('127.0.0.1', 5002)]

--- 模拟ECU1故障（关闭5001端口）---
[Client] 切换到备用实例 ('127.0.0.1', 5002)
调用结果: 成功
当前使用实例: ('127.0.0.1', 5002)

通过条件：客户端在首选实例（5001端口）不可达后，自动切换至备用实例（5002端口）并成功完成调用。若输出中“调用结果: 成功”且“当前使用实例”指向5002端口，则冗余机制验证通过。

关键说明

• SD服务器维护实例表时，以instance_id为唯一键，后注册的同ID实例会覆盖旧实例，模拟实例重启场景。
• 客户端故障切换采用顺序轮询策略，每次调用失败后移动current_idx指向下一候选。
• 本案例使用UDP模拟SOME/IP SD报文，实际部署中需替换为SOME/IP协议栈（如vSOME/IP）的SD实现，但冗余发现与切换逻辑保持一致。