2. DDS发现机制基础实现

任务目标

实现一个最小化的DDS发现机制：两个参与者（Participant）通过UDP组播发送SPDP（Simple Participant Discovery Protocol）报文，相互发现对方的存在，并建立可用的通信端点信息交换。本案例聚焦于发现阶段，不涉及后续的RTPS可靠传输或QoS策略。

环境与依赖

使用Python 3.8+，仅依赖标准库（socket, threading, json, time）。无需安装第三方DDS库，通过UDP组播模拟DDS发现的核心机制。

实现原理

DDS发现的核心是SPDP：每个参与者启动时向预定义的组播地址（此处用 239.255.0.1:7400）周期性地发送自身信息（参与者GUID、IP、端口），同时监听该组播地址接收其他参与者的宣告。当收到新的参与者信息时，将其记录到本地参与者表中，完成发现。

完整代码

import socket
import threading
import json
import time
import uuid

MCAST_GRP = '239.255.0.1'
MCAST_PORT = 7400
ANNOUNCE_INTERVAL = 2.0  # 宣告周期（秒）
LISTEN_TIMEOUT = 0.5     # 监听超时（秒）

class DDSDiscoveryParticipant:
    def __init__(self, name: str):
        self.name = name
        self.guid = str(uuid.uuid4())  # 全局唯一标识
        self.port = 7400 + int(self.guid[:2], 16) % 100  # 模拟数据端口
        self.discovered = {}  # guid -> {name, port, last_seen}
        self.running = False

        # 创建组播套接字
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM, socket.IPPROTO_UDP)
        self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        self.sock.bind(('', MCAST_PORT))
        mreq = socket.inet_aton(MCAST_GRP) + socket.inet_aton('0.0.0.0')
        self.sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq)
        self.sock.settimeout(LISTEN_TIMEOUT)

    def _build_announcement(self) -> bytes:
        """构造SPDP宣告报文"""
        info = {
            'guid': self.guid,
            'name': self.name,
            'port': self.port,
            'timestamp': time.time()
        }
        return json.dumps(info).encode('utf-8')

    def _send_announcement(self):
        """发送宣告到组播组"""
        data = self._build_announcement()
        self.sock.sendto(data, (MCAST_GRP, MCAST_PORT))

    def _listen_once(self):
        """监听一次组播消息，处理收到的宣告"""
        try:
            data, addr = self.sock.recvfrom(4096)
        except socket.timeout:
            return

        try:
            info = json.loads(data.decode('utf-8'))
            guid = info['guid']
            if guid != self.guid:  # 忽略自己的宣告
                self.discovered[guid] = {
                    'name': info['name'],
                    'port': info['port'],
                    'last_seen': time.time()
                }
                print(f"[{self.name}] 发现参与者: {info['name']} (GUID: {guid[:8]}...)")
        except (json.JSONDecodeError, KeyError):
            pass  # 忽略无效报文

    def start(self):
        """启动发现线程"""
        self.running = True
        thread = threading.Thread(target=self._run, daemon=True)
        thread.start()

    def _run(self):
        """主循环：周期宣告 + 持续监听"""
        while self.running:
            self._send_announcement()
            # 在宣告间隔内多次监听
            end_time = time.time() + ANNOUNCE_INTERVAL
            while time.time() < end_time and self.running:
                self._listen_once()

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

    def get_discovered_count(self) -> int:
        return len(self.discovered)

def run_demo():
    """演示：启动两个参与者，验证相互发现"""
    p1 = DDSDiscoveryParticipant("Participant-A")
    p2 = DDSDiscoveryParticipant("Participant-B")

    p1.start()
    p2.start()

    # 等待发现完成（最多5秒）
    time.sleep(5)

    print(f"\n=== 发现结果 ===")
    print(f"{p1.name} 发现 {p1.get_discovered_count()} 个参与者")
    print(f"{p2.name} 发现 {p2.get_discovered_count()} 个参与者")

    # 验证：双方应互相发现
    assert p1.get_discovered_count() >= 1, "Participant-A 未发现任何参与者"
    assert p2.get_discovered_count() >= 1, "Participant-B 未发现任何参与者"

    # 验证发现的是对方
    p1_discovered_names = [v['name'] for v in p1.discovered.values()]
    p2_discovered_names = [v['name'] for v in p2.discovered.values()]
    assert "Participant-B" in p1_discovered_names, "A 未发现 B"
    assert "Participant-A" in p2_discovered_names, "B 未发现 A"

    print("✓ 验证通过：双方成功相互发现")

    p1.stop()
    p2.stop()

if __name__ == '__main__':
    run_demo()

运行方式

python dds_discovery.py

验证方法与判定标准

运行上述脚本，观察输出：

  1. 两个参与者启动后，控制台应打印类似 [Participant-A] 发现参与者: Participant-B (GUID: 3f2a1c...) 的日志，表明收到了对方的SPDP宣告。
  2. 最终输出应包含 === 发现结果 === 以及双方发现数量均为1。
  3. 程序以 ✓ 验证通过：双方成功相互发现 结束，退出码为0，说明断言全部通过。

若在真实网络环境中运行，需确保本机防火墙允许UDP组播通信（目标端口7400）。

关键说明

  • 本实现使用JSON作为SPDP报文的序列化格式，实际DDS使用CDR编码的二进制格式，但发现逻辑等价。
  • 参与者通过SO_REUSEADDR允许多个实例绑定同一端口，这是组播接收的必要条件。
  • 发现结果保存在内存字典中，实际DDS实现会在此基础上建立端到端的通信连接。
← 上一章 📖 返回目录 下一章 →