1. SOME/IP服务发现基础实现

任务说明

本案例实现SOME/IP服务发现（Service Discovery）的最小闭环：一个服务提供者（Server）通过SD报文宣告自身服务，一个消费者（Client）通过查找报文发现该服务并完成订阅。代码基于Python + vsomeip库实现，不依赖真实ECU环境，可在本机直接运行验证。

环境依赖

• Python 3.8+
• 安装vsomeip库：pip install vsomeip（Linux环境需先安装Boost与vsomeip原生库，Windows可跳过本案例）

完整实现代码

import vsomeip
import time
import threading

# 服务ID与实例ID（SOME/IP标准定义）
SERVICE_ID = 0x1234
INSTANCE_ID = 0x5678
EVENTGROUP_ID = 0x0001
EVENT_ID = 0x8001

# 服务提供者（Server）端
class ServiceProvider:
    def __init__(self):
        self.app = vsomeip.Application("provider")
        self.app.init()
        self.app.register_message_handler(vsomeip.message_type_e.MT_REQUEST, self.on_request)
        self.app.register_availability_handler(self.on_availability)
        self.app.register_state_handler(self.on_state)
        self.offered = False

    def on_state(self, state):
        if state == vsomeip.application_state_e.APP_STATE_REGISTERED:
            self.app.offer_service(SERVICE_ID, INSTANCE_ID)
            self.offered = True
            print("[Provider] Service offered: 0x1234/0x5678")

    def on_availability(self, service_id, instance_id, available):
        pass  # 服务端不需要监听自身可用性

    def on_request(self, message):
        # 处理消费者发来的请求（此处为简单回显）
        response = self.app.create_response(message)
        response.set_payload(message.get_payload())
        self.app.send(response)
        print("[Provider] Request handled")

    def start(self):
        self.app.start()

# 消费者（Client）端
class ServiceConsumer:
    def __init__(self):
        self.app = vsomeip.Application("consumer")
        self.app.init()
        self.app.register_availability_handler(self.on_availability)
        self.app.register_state_handler(self.on_state)
        self.service_found = False

    def on_state(self, state):
        if state == vsomeip.application_state_e.APP_STATE_REGISTERED:
            # 查找服务
            self.app.request_service(SERVICE_ID, INSTANCE_ID)
            print("[Consumer] Requesting service 0x1234/0x5678")

    def on_availability(self, service_id, instance_id, available):
        if available:
            self.service_found = True
            print(f"[Consumer] Service found: 0x{service_id:04x}/0x{instance_id:04x}")
            # 订阅事件组
            self.app.request_event(SERVICE_ID, INSTANCE_ID, EVENT_ID, EVENTGROUP_ID)
            self.app.subscribe(SERVICE_ID, INSTANCE_ID, EVENTGROUP_ID)
            print("[Consumer] Subscribed to event group")
        else:
            self.service_found = False
            print("[Consumer] Service lost")

    def start(self):
        self.app.start()

if __name__ == "__main__":
    import sys
    if len(sys.argv) < 2:
        print("Usage: python sd_demo.py [server|client]")
        sys.exit(1)

    role = sys.argv[1]
    if role == "server":
        provider = ServiceProvider()
        provider.start()
    elif role == "client":
        consumer = ServiceConsumer()
        consumer.start()
    else:
        print("Invalid role. Use 'server' or 'client'.")
        sys.exit(1)

运行与验证

本案例需要两个终端分别启动服务端与客户端：

# 终端1：启动服务提供者
python sd_demo.py server

# 终端2：启动消费者（等待1~2秒让服务端先完成注册）
python sd_demo.py client

通过条件：

• 终端1输出 [Provider] Service offered: 0x1234/0x5678，表明服务已注册到SD。
• 终端2输出 [Consumer] Service found: 0x1234/0x5678，表明消费者通过SD报文成功发现服务。
• 终端2继续输出 [Consumer] Subscribed to event group，表明事件组订阅完成。

若两端均无报错且上述日志按序出现，则说明SOME/IP服务发现基本流程（提供者注册 + 消费者查找）实现成功。

关键说明

• vsomeip的SD报文（OfferService/FindService）由库内部自动封装发送，本案例代码聚焦于应用层注册与查找逻辑。
• 服务端与客户端需运行在同一网络命名空间（本机回环即可），vsomeip默认使用回环地址通信。
• 若需跨设备运行，需在vsomeip的JSON配置文件中指定网卡IP，本案例不展开。