1. SOME/IP协议报文结构解析与手工构造

任务说明

实现一个最小可运行的SOME/IP服务发现（SD）模块，完成以下闭环：

服务端（Service Provider）启动时发送Service Offered报文，宣告服务上线；
客户端（Service Consumer）接收并解析该报文，识别服务实例已上线；
服务端退出时发送Service Stopped报文，客户端接收后识别服务下线。

本案例聚焦SD协议中最核心的报文格式与状态通知机制，不涉及TCP连接建立、事件订阅等扩展功能。

环境与依赖

Python 3.8+，仅使用标准库（socket, struct, time）
无需第三方库，单文件即可运行

协议设计（最小实现）

SOME/IP SD报文基于UDP传输，默认端口30490。本案例使用以下精简报文结构（与AUTOSAR SD规范兼容的字段子集）：

Message ID（4字节）：0xFFFF8100 表示SD报文
Request ID（4字节）：Client ID（2字节）+ Session ID（2字节）
Protocol Version（1字节）：0x01
Interface Version（1字节）：0x01
Message Type（1字节）：0x02（Notification，SD报文固定类型）
Return Code（1字节）：0x00
Length（4字节）：后续Entry数组与Option数组的总字节数
Entries：每个Entry 16字节，包含Type（1字节）、Service ID（2字节）、Instance ID（2字节）、Major Version（1字节）、TTL（4字节）、Minor Version（4字节）

Entry Type定义：0x00 = Service Offered（上线），0x01 = Service Stopped（下线）。

完整实现代码

import socket
import struct
import time
import threading

# SD报文常量
SD_MESSAGE_ID = 0xFFFF8100
SD_PORT = 30490
PROTOCOL_VERSION = 0x01
INTERFACE_VERSION = 0x01
MESSAGE_TYPE_NOTIFICATION = 0x02
RETURN_CODE_OK = 0x00

# Entry Type
ENTRY_SERVICE_OFFERED = 0x00
ENTRY_SERVICE_STOPPED = 0x01

# 本案例使用的服务标识
SERVICE_ID = 0x1234
INSTANCE_ID = 0x5678
MAJOR_VERSION = 0x01
MINOR_VERSION = 0x00000001

def build_sd_packet(entry_type, ttl):
    """构建一个包含单个Entry的SD报文"""
    # Entry字段
    entry = struct.pack('!BHHBIB', entry_type, SERVICE_ID, INSTANCE_ID,
                        MAJOR_VERSION, ttl, 0)  # 最后1字节为保留位
    entry += struct.pack('!I', MINOR_VERSION)

    # 计算Length：Entry数组长度 + Option数组长度（本案例无Option，填0）
    entries_length = len(entry)
    options_length = 0
    total_length = entries_length + options_length

    # SD报文头
    header = struct.pack('!IHHBBBB', SD_MESSAGE_ID, 0x0001, 0x0001,
                         PROTOCOL_VERSION, INTERFACE_VERSION,
                         MESSAGE_TYPE_NOTIFICATION, RETURN_CODE_OK)
    header += struct.pack('!I', total_length)

    return header + entry

def parse_sd_packet(data):
    """解析SD报文，返回(entry_type, service_id, instance_id, ttl)或None"""
    if len(data) < 16:
        return None

    message_id, client_id, session_id, proto_ver, iface_ver, msg_type, ret_code = \
        struct.unpack('!IHHBBBB', data[:12])

    if message_id != SD_MESSAGE_ID:
        return None

    length = struct.unpack('!I', data[12:16])[0]
    if len(data) < 16 + length:
        return None

    # 解析第一个Entry（本案例只发送单个Entry）
    entry_data = data[16:32]
    if len(entry_data) < 16:
        return None

    entry_type, service_id, instance_id, major_ver, ttl = \
        struct.unpack('!BHHBI', entry_data[:10])

    return entry_type, service_id, instance_id, ttl

def run_service_provider(duration=5):
    """服务端：先发送上线报文，等待duration秒后发送下线报文"""
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
    sock.bind(('', 0))  # 使用临时端口发送

    # 发送上线通知
    offer_packet = build_sd_packet(ENTRY_SERVICE_OFFERED, ttl=5)
    sock.sendto(offer_packet, ('127.0.0.1', SD_PORT))
    print(f'[Provider] Service Offered sent: Service 0x{SERVICE_ID:04X}, Instance 0x{INSTANCE_ID:04X}')

    time.sleep(duration)

    # 发送下线通知
    stop_packet = build_sd_packet(ENTRY_SERVICE_STOPPED, ttl=0)
    sock.sendto(stop_packet, ('127.0.0.1', SD_PORT))
    print(f'[Provider] Service Stopped sent: Service 0x{SERVICE_ID:04X}, Instance 0x{INSTANCE_ID:04X}')

    sock.close()

def run_service_consumer(timeout=8):
    """客户端：监听SD报文，打印上线/下线事件"""
    sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
    sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    sock.bind(('127.0.0.1', SD_PORT))
    sock.settimeout(timeout)

    print(f'[Consumer] Listening on port {SD_PORT}...')

    try:
        while True:
            data, addr = sock.recvfrom(4096)
            result = parse_sd_packet(data)
            if result is None:
                continue

            entry_type, service_id, instance_id, ttl = result
            if entry_type == ENTRY_SERVICE_OFFERED:
                print(f'[Consumer] Service ONLINE: 0x{service_id:04X}/0x{instance_id:04X}, TTL={ttl}s')
            elif entry_type == ENTRY_SERVICE_STOPPED:
                print(f'[Consumer] Service OFFLINE: 0x{service_id:04X}/0x{instance_id:04X}')
    except socket.timeout:
        print('[Consumer] Timeout reached, exiting.')

    sock.close()

if __name__ == '__main__':
    # 启动消费者线程
    consumer_thread = threading.Thread(target=run_service_consumer, daemon=True)
    consumer_thread.start()

    time.sleep(0.5)  # 确保消费者已绑定端口

    # 运行服务提供者（5秒后自动发送下线）
    run_service_provider(duration=5)

    consumer_thread.join(timeout=10)

运行方式

python sd_basic.py

验证方法与通过标准

程序启动后，消费者线程先监听UDP 30490端口；
服务端发送上线报文，消费者应打印：Service ONLINE: 0x1234/0x5678, TTL=5s；
5秒后服务端发送下线报文，消费者应打印：Service OFFLINE: 0x1234/0x5678；
消费者在8秒超时后自动退出，程序结束。

若控制台按上述顺序输出两条状态通知，且无异常报错，即判定案例通过。可通过修改run_service_provider(duration=...)参数调整上下线间隔时间。

常见问题

端口被占用：若30490端口已被其他程序占用，消费者绑定会失败，可修改SD_PORT常量并保持两端一致。
收不到报文：确认服务端发送地址为127.0.0.1且消费者已成功绑定；若使用多网卡环境，可改用广播地址255.255.255.255发送。
TTL过期处理：本案例未实现TTL定时刷新机制，实际工程中客户端需在TTL超时后主动判定服务下线，此处为聚焦SD报文收发而省略。

📚 课程目录