2. 主题发现与匹配

任务目标

在DDS（Data Distribution Service）中，主题（Topic）是发布者与订阅者通信的桥梁。本案例通过构建一个最小化的发布-订阅示例，观察以下三种场景下的主题发现与匹配行为：

  1. 主题名相同、数据类型相同：应成功匹配并完成通信；
  2. 主题名不同、数据类型相同：不应匹配，订阅者收不到数据；
  3. 主题名相同、数据类型不匹配：不应匹配，订阅者收不到数据。

通过观察匹配结果，理解DDS主题匹配的核心规则：主题名与数据类型必须同时一致。

环境与依赖

本案例基于 Fast DDS（eProsima Fast DDS）实现，使用Python绑定。需安装：

pip install fastdds

完整实现

以下代码创建两个发布者（分别发布到 TopicA 和 TopicB）与两个订阅者（分别订阅 TopicA 和 TopicA_TypeMismatch），覆盖上述三种场景。为简化演示，数据类型使用Fast DDS内置的 String 类型，以及一个自定义的 IntTopic 类型用于类型不匹配测试。

import time
import threading
from dataclasses import dataclass

import fastdds

# ---------- 自定义数据类型（用于类型不匹配场景） ----------
@dataclass
class IntTopic:
    value: int

# ---------- 监听器：跟踪匹配状态 ----------
class PubListener(fastdds.PublisherListener):
    def __init__(self):
        super().__init__()
        self.matched = 0

    def on_publication_matched(self, pub, info):
        if info.status == fastdds.MatchingStatus.MATCHED_MATCHING:
            self.matched += 1
            print(f"[Publisher] 发现匹配的订阅者，当前匹配数: {self.matched}")
        elif info.status == fastdds.MatchingStatus.REMOVED_MATCHING:
            self.matched -= 1
            print(f"[Publisher] 订阅者断开，当前匹配数: {self.matched}")

class SubListener(fastdds.SubscriberListener):
    def __init__(self):
        super().__init__()
        self.matched = 0
        self.received = 0

    def on_subscription_matched(self, sub, info):
        if info.status == fastdds.MatchingStatus.MATCHED_MATCHING:
            self.matched += 1
            print(f"[Subscriber] 发现匹配的发布者，当前匹配数: {self.matched}")
        elif info.status == fastdds.MatchingStatus.REMOVED_MATCHING:
            self.matched -= 1
            print(f"[Subscriber] 发布者断开，当前匹配数: {self.matched}")

    def on_data_available(self, sub):
        reader = sub.get_datareader()
        if reader.take_next_sample():
            self.received += 1
            print(f"[Subscriber] 收到数据，累计: {self.received}")

# ---------- 创建参与者 ----------
def create_participant(domain_id, name):
    dp = fastdds.DomainParticipantFactory.get_instance().create_participant(
        domain_id,
        fastdds.PARTICIPANT_QOS_DEFAULT,
        None,
        fastdds.StatusMask.all()
    )
    dp.set_qos(fastdds.DomainParticipantQos())
    return dp

# ---------- 场景1：主题名相同、类型相同（String） ----------
def scenario_match():
    print("\n===== 场景1: TopicA / String 发布者 + TopicA / String 订阅者 =====")
    participant = create_participant(0, "participant_match")

    # 发布者
    pub_listener = PubListener()
    publisher = participant.create_publisher(fastdds.PUBLISHER_QOS_DEFAULT, pub_listener)
    topic_pub = participant.create_topic("TopicA", "std::string", fastdds.TOPIC_QOS_DEFAULT)
    writer = publisher.create_datawriter(topic_pub, fastdds.DATAWRITER_QOS_DEFAULT)

    # 订阅者
    sub_listener = SubListener()
    subscriber = participant.create_subscriber(fastdds.SUBSCRIBER_QOS_DEFAULT, sub_listener)
    topic_sub = participant.create_topic("TopicA", "std::string", fastdds.TOPIC_QOS_DEFAULT)
    reader = subscriber.create_datareader(topic_sub, fastdds.DATAREADER_QOS_DEFAULT)

    time.sleep(1)  # 等待发现

    # 发布数据
    msg = "Hello DDS"
    writer.write(msg)
    time.sleep(1)

    print(f"匹配结果: 发布者匹配数={pub_listener.matched}, 订阅者匹配数={sub_listener.matched}, 收到数据数={sub_listener.received}")
    assert pub_listener.matched == 1 and sub_listener.matched == 1 and sub_listener.received == 1, "场景1应成功匹配并收到数据"

    # 清理
    participant.delete_contained_entities()
    fastdds.DomainParticipantFactory.get_instance().delete_participant(participant)

# ---------- 场景2：主题名不同 ----------
def scenario_topic_mismatch():
    print("\n===== 场景2: TopicA / String 发布者 + TopicB / String 订阅者 =====")
    participant = create_participant(0, "participant_topic_mismatch")

    pub_listener = PubListener()
    publisher = participant.create_publisher(fastdds.PUBLISHER_QOS_DEFAULT, pub_listener)
    topic_pub = participant.create_topic("TopicA", "std::string", fastdds.TOPIC_QOS_DEFAULT)
    writer = publisher.create_datawriter(topic_pub, fastdds.DATAWRITER_QOS_DEFAULT)

    sub_listener = SubListener()
    subscriber = participant.create_subscriber(fastdds.SUBSCRIBER_QOS_DEFAULT, sub_listener)
    topic_sub = participant.create_topic("TopicB", "std::string", fastdds.TOPIC_QOS_DEFAULT)
    reader = subscriber.create_datareader(topic_sub, fastdds.DATAREADER_QOS_DEFAULT)

    time.sleep(1)

    writer.write("Hello DDS")
    time.sleep(1)

    print(f"匹配结果: 发布者匹配数={pub_listener.matched}, 订阅者匹配数={sub_listener.matched}, 收到数据数={sub_listener.received}")
    assert pub_listener.matched == 0 and sub_listener.matched == 0 and sub_listener.received == 0, "场景2不应匹配"

    participant.delete_contained_entities()
    fastdds.DomainParticipantFactory.get_instance().delete_participant(participant)

# ---------- 场景3：类型不匹配 ----------
def scenario_type_mismatch():
    print("\n===== 场景3: TopicA / String 发布者 + TopicA / IntTopic 订阅者 =====")
    participant = create_participant(0, "participant_type_mismatch")

    # 注册自定义类型
    type_support = fastdds.TypeSupport()
    type_support.set_name("IntTopic")
    participant.register_type(type_support)

    pub_listener = PubListener()
    publisher = participant.create_publisher(fastdds.PUBLISHER_QOS_DEFAULT, pub_listener)
    topic_pub = participant.create_topic("TopicA", "std::string", fastdds.TOPIC_QOS_DEFAULT)
    writer = publisher.create_datawriter(topic_pub, fastdds.DATAWRITER_QOS_DEFAULT)

    sub_listener = SubListener()
    subscriber = participant.create_subscriber(fastdds.SUBSCRIBER_QOS_DEFAULT, sub_listener)
    topic_sub = participant.create_topic("TopicA", "IntTopic", fastdds.TOPIC_QOS_DEFAULT)
    reader = subscriber.create_datareader(topic_sub, fastdds.DATAREADER_QOS_DEFAULT)

    time.sleep(1)

    writer.write("Hello DDS")
    time.sleep(1)

    print(f"匹配结果: 发布者匹配数={pub_listener.matched}, 订阅者匹配数={sub_listener.matched}, 收到数据数={sub_listener.received}")
    assert pub_listener.matched == 0 and sub_listener.matched == 0 and sub_listener.received == 0, "场景3不应匹配"

    participant.delete_contained_entities()
    fastdds.DomainParticipantFactory.get_instance().delete_participant(participant)

# ---------- 主入口 ----------
if __name__ == "__main__":
    scenario_match()
    scenario_topic_mismatch()
    scenario_type_mismatch()
    print("\n===== 全部场景执行完毕 =====")

运行与验证

直接运行上述Python脚本：

python topic_matching_demo.py

程序依次执行三个场景，每个场景通过assert自动验证匹配结果。预期输出如下（实际输出可能因时序略有差异，但匹配数与收数数应一致）：

===== 场景1: TopicA / String 发布者 + TopicA / String 订阅者 =====
[Publisher] 发现匹配的订阅者，当前匹配数: 1
[Subscriber] 发现匹配的发布者，当前匹配数: 1
[Subscriber] 收到数据，累计: 1
匹配结果: 发布者匹配数=1, 订阅者匹配数=1, 收到数据数=1

===== 场景2: TopicA / String 发布者 + TopicB / String 订阅者 =====
匹配结果: 发布者匹配数=0, 订阅者匹配数=0, 收到数据数=0

===== 场景3: TopicA / String 发布者 + TopicA / IntTopic 订阅者 =====
匹配结果: 发布者匹配数=0, 订阅者匹配数=0, 收到数据数=0

===== 全部场景执行完毕 =====

通过条件：脚本无异常退出，且三个场景的assert全部通过，即场景1匹配数为1且收到1条数据，场景2和场景3匹配数均为0且未收到数据。

关键说明

  • 场景2中，发布者与订阅者处于同一Domain（ID=0），但主题名不同，DDS不会建立连接，因此on_publication_matched与on_subscription_matched回调均不会触发。
  • 场景3中，主题名相同但数据类型不同（std::string vs IntTopic），DDS的类型一致性检查会阻止匹配。注意自定义类型需先通过register_type注册到参与者。
  • 匹配回调中的info.status用于区分匹配建立（MATCHED_MATCHING）与断开（REMOVED_MATCHING），本案例仅关注建立事件。
← 上一章 📖 返回目录 下一章 →