2. 双通道冗余通信

任务目标

实现一个基于双独立网络通道的SOME/IP通信节点：正常情况下通过主通道收发消息，当主通道发生故障（如链路断开、超时无响应）时，自动切换至备用通道继续通信，切换过程对应用层透明，保证消息传输不中断。

实现方案

采用通道抽象层 + 故障检测 + 自动切换的最小可复现设计。每个通道封装独立的UDP Socket（模拟两个物理网络），上层通过统一接口收发SOME/IP消息。故障检测基于周期性心跳超时判断，主通道连续3次心跳超时即触发切换。切换后备用通道接管收发，应用层无需感知通道变化。

关键完整代码

import socket
import threading
import time
import struct

# SOME/IP 消息头 (简化: Message ID + Length + Payload)
SOMEIP_HEADER = struct.Struct("!II")
HEARTBEAT_ID = 0x10000001
DATA_ID = 0x10000002
HEARTBEAT_INTERVAL = 1.0
HEARTBEAT_TIMEOUT = 3.0  # 连续3次超时判定故障

class Channel:
    """单条网络通道封装"""
    def __init__(self, local_ip, local_port, remote_ip, remote_port, name):
        self.name = name
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.sock.bind((local_ip, local_port))
        self.sock.settimeout(0.5)
        self.remote = (remote_ip, remote_port)
        self.alive = True
        self._lock = threading.Lock()

    def send(self, msg_id, payload):
        data = SOMEIP_HEADER.pack(msg_id, len(payload)) + payload
        with self._lock:
            self.sock.sendto(data, self.remote)

    def recv(self):
        try:
            data, _ = self.sock.recvfrom(4096)
            msg_id, length = SOMEIP_HEADER.unpack(data[:8])
            return msg_id, data[8:8+length]
        except socket.timeout:
            return None, None

    def close(self):
        self.alive = False
        self.sock.close()

class DualChannelNode:
    """双通道冗余通信节点"""
    def __init__(self, primary: Channel, backup: Channel):
        self.primary = primary
        self.backup = backup
        self.active = primary  # 当前活动通道
        self.heartbeat_failures = 0
        self.running = True
        self._recv_thread = threading.Thread(target=self._recv_loop, daemon=True)
        self._hb_thread = threading.Thread(target=self._heartbeat_loop, daemon=True)

    def start(self):
        self._recv_thread.start()
        self._hb_thread.start()

    def send_message(self, payload):
        """应用层发送接口，自动选择活动通道"""
        self.active.send(DATA_ID, payload)

    def _recv_loop(self):
        while self.running:
            msg_id, payload = self.active.recv()
            if msg_id == HEARTBEAT_ID:
                self.heartbeat_failures = 0  # 收到心跳，重置计数
            elif msg_id == DATA_ID:
                print(f"[{self.active.name}] 收到数据: {payload.decode()}")

    def _heartbeat_loop(self):
        while self.running:
            time.sleep(HEARTBEAT_INTERVAL)
            self.active.send(HEARTBEAT_ID, b"ping")
            # 等待接收线程更新心跳状态
            time.sleep(0.1)
            self.heartbeat_failures += 1
            if self.heartbeat_failures >= HEARTBEAT_TIMEOUT:
                self._switch_channel()

    def _switch_channel(self):
        """切换活动通道到备用通道"""
        old = self.active
        self.active = self.backup if self.active == self.primary else self.primary
        self.heartbeat_failures = 0
        print(f"[冗余切换] {old.name} 故障 → 切换至 {self.active.name}")

    def stop(self):
        self.running = False
        self.primary.close()
        self.backup.close()

# ========== 最小验证场景 ==========
if __name__ == "__main__":
    # 节点A: 主通道192.168.1.10, 备用通道10.0.0.10
    ch_a1 = Channel("192.168.1.10", 5000, "192.168.1.20", 5000, "主通道A")
    ch_a2 = Channel("10.0.0.10", 5000, "10.0.0.20", 5000, "备用通道A")
    node_a = DualChannelNode(ch_a1, ch_a2)

    # 节点B: 主通道192.168.1.20, 备用通道10.0.0.20
    ch_b1 = Channel("192.168.1.20", 5000, "192.168.1.10", 5000, "主通道B")
    ch_b2 = Channel("10.0.0.20", 5000, "10.0.0.10", 5000, "备用通道B")
    node_b = DualChannelNode(ch_b1, ch_b2)

    node_a.start()
    node_b.start()

    # 模拟应用发送数据
    for i in range(10):
        node_a.send_message(f"msg-{i}".encode())
        time.sleep(1)
        if i == 3:
            # 模拟主通道故障：关闭节点A的主通道socket
            print("\n[故障注入] 关闭节点A主通道...")
            ch_a1.close()
            ch_a1.alive = False

    time.sleep(5)
    node_a.stop()
    node_b.stop()
    print("\n验证完成")

运行与验证

环境要求：Python 3.6+，无需第三方库。需在支持多IP地址的机器上运行（或使用127.0.0.1的不同端口模拟两个通道）。

运行方式：

# 若本机有多个IP，直接运行；否则将代码中IP改为127.0.0.1，端口区分
python dual_channel_someip.py

验证方法：

  1. 观察前3条消息（msg-0~msg-2）通过主通道正常收发，控制台打印[主通道A] 收到数据。
  2. 第4次循环时注入主通道故障，随后应看到[冗余切换] 主通道A 故障 → 切换至 备用通道A。
  3. 后续消息（msg-4~msg-9）应通过备用通道继续收发，控制台打印[备用通道A] 收到数据，证明切换后通信未中断。

通过判定标准：故障注入后，备用通道成功接管收发，且后续所有消息均能到达对端节点，无消息丢失。

关键设计说明

  • 心跳超时阈值设为3秒（3次心跳间隔），可根据实际网络延迟调整HEARTBEAT_INTERVAL与HEARTBEAT_TIMEOUT。
  • 切换逻辑采用主动检测+立即切换，无回切机制；实际工程可增加主通道恢复检测后自动回切。
  • 本实现聚焦双通道冗余核心机制，未包含SOME/IP完整序列化、服务发现等协议细节，但消息头结构可扩展。
← 上一章 📖 返回目录 下一章 →