Matter mDNS 服务发现指南：深入解析 connectedhomeip 中的 MdnsDiscovery 工具

Matter mDNS 服务发现指南：深入解析 connectedhomeip 中的 MdnsDiscovery 工具

【免费下载链接】connectedhomeipMatter (formerly Project CHIP) creates more connections between more objects, simplifying development for manufacturers and increasing compatibility for consumers, guided by the Connectivity Standards Alliance.项目地址: https://gitcode.com/GitHub_Trending/co/connectedhomeip

导读

MdnsDiscovery是 Matter（connectedhomeip）仓库中面向 mDNS 服务发现的核心 Python 工具类，用于在本地网络（IPv6）上发现 Matter 设备的四种核心角色服务——commissioner（管理员）、commissionable node（可配网节点）、operational node（已入网节点）与 Thread border router（边界路由器），底层基于python-zeroconf库实现异步多播 DNS 查询。本文将以 README.md 为主线，结合仓库内 mdns_discovery.py 及其配套模块的源码实现，系统讲解该工具的设计结构、四大服务发现方法、DNS 记录（PTR/SRV/TXT/A/AAAA）直查能力、发现流程的内部机制，以及配套的 mDNS 值校验断言与网络工具函数。读完本文，你将掌握在 Matter 集成测试与日常调试中利用MdnsDiscovery完成服务发现、记录查询与结果校验的完整实战方案。

背景：Matter 的 mDNS 服务发现为何重要

Matter 设备在本地网络上的互相发现依赖mDNS（多播 DNS）与 DNS-SD（DNS Service Discovery）。设备入网后会在本地网络宣告自身的服务，控制器（controller）同样通过 mDNS 找到这些服务。MdnsDiscovery正是为这类发现场景提供统一异步接口的工具：它既支持高层语义的服务查询（如"找所有可配网设备"），也支持直接访问低层 DNS 记录（PTR、SRV、TXT、A、AAAA），满足测试场景中对原始记录验证的需求。

从源码看，该工具被广泛用于 Matter 的 Python 集成测试用例中，例如：

• TC_SC_4_1.py 使用MdnsDiscovery().get_ptr_records()对 Discriminator Subtype 发起 PTR 记录查询，验证开放配网窗口期间公告的 subtype 记录；
• commissioning.py 中的_is_device_operational_via_dnssd()与_is_device_commissionable_via_dnssd()分别调用get_operational_services()和get_commissionable_services()，在建立 CASE 会话前先通过 DNS-SD 判断设备当前状态，避免不必要的超时等待。

这些实际用法证明该模块不是孤立的示例代码，而是 Matter 测试基础设施中可复用的发现能力组件。

模块结构与目录说明

MdnsDiscovery位于 src/python_testing/mdns_discovery，目录结构如下：

📁 mdns_discovery/ ├── 📁data_classes/ # 服务信息与查询结果的容器（MdnsServiceInfo / PtrRecord / AaaaRecord 等） ├── 📁enums/ # 服务类型等枚举定义（MdnsServiceType） ├── 📁service_listeners/ # 发现会话期间使用的服务监听器（MdnsServiceListener） ├── 📁tests/ # 断言函数及其他方法的单元测试（test_asserts.py） ├── 📁utils/ # 工具函数：IPv6 过滤与断言等（network.py / asserts.py） ├── 📄mdns_async_service_info.py # 支持查询指定 mDNS 记录类型的封装（MdnsAsyncServiceInfo / AddressResolverIPv6） └── 📄mdns_discovery.py # mDNS 发现操作的主入口（MdnsDiscovery）

各模块职责清晰：

• enums/mdns_service_type.py定义了 Matter 的四种核心 mDNS 服务类型（MdnsServiceType枚举）；
• data_classes定义了三种可序列化数据容器：MdnsServiceInfo（完整服务信息）、PtrRecord（PTR 记录）与AaaaRecord（IPv6 地址记录），均继承自JsonSerializable以支持 JSON 输出（见 mdns_service_info.py）；
• service_listeners/mdns_service_listener.py实现了一个内部asyncio.Event驱动的服务监听器，用于在记录查询前同步等待服务 add/update 事件；
• mdns_async_service_info.py对zeroconf.ServiceInfo做了子类化，重写async_request以禁用缓存、强制发起全新网络查询，并提供仅解析 AAAA（IPv6）记录的AddressResolverIPv6。

核心 API 总览

MdnsDiscovery类在初始化时（mdns_discovery.py）会通过get_host_ipv6_addresses()自动采集本机全部可用的 IPv6 接口地址（含 link-local 时附带%接口名作用域），并以_discovered_services字典保存发现结果、以asyncio.Event同步发现过程。其公开 API 可分为三大类：

1. 服务发现方法（高层语义查询）

四个 Matter 角色方法实现高度一致：调用discover(service_types=[...], query_service=True, append_results=True, ...)，随后从self._discovered_services按服务类型取值返回list[MdnsServiceInfo]。这四种服务类型值定义于 mdns_service_type.py，与 Matter 规范中四个角色的公告类型一一对应。

2. 记录查询方法（低层 DNS 记录直查）

其中get_srv_record与get_txt_record的内部实现仅改变query_record_types参数（{_TYPE_SRV}或{_TYPE_TXT}），统一走_query_service_info()私有方法；get_quada_records则使用AddressResolverIPv6单独解析 AAAA 记录。

3. 服务类型发现方法

get_commissionable_subtypes()的实现（mdns_discovery.py）先调用get_all_service_types()，再筛选以_开头且包含._sub._matterc._udp.local.的服务类型，返回的正是_L\d+（长判别器）、_S\d+（短判别器）、_V\d+（厂商）、_T\d+（设备类型）等配网 subtype。

快速上手：两个最小可运行示例

示例一：发现 operational 节点

import asyncio from mdns_discovery.mdns_discovery import MdnsDiscovery async def main(): mdns = MdnsDiscovery() # 发现 operational 节点（_matter._tcp.local.） services = await mdns.get_operational_services() # 打印基本信息 for service in services: print(f"Instance: {service.instance_name}") print(f"Addresses: {service.addresses}") print(f"Port: {service.port}") print("---") asyncio.run(main())

示例二：查询指定服务实例的 SRV 记录

import asyncio from mdns_discovery.mdns_discovery import MdnsDiscovery async def main(): mdns = MdnsDiscovery() service_name = 'B7322C948581262F-0000000012344321._matter._tcp.local.' # 查询 SRV 记录 srv_record = await mdns.get_srv_record( service_name=service_name, service_type=MdnsServiceType.OPERATIONAL.value, log_output=True ) # 打印主机名 print(f"Hostname: {srv_record.hostname}") asyncio.run(main())

注意：operational 实例名遵循{16位压缩Fabric ID}-{16位节点ID}格式（如B7322C948581262F-0000000012344321），这一点在 asserts.py 中的assert_valid_operational_instance_name有严格校验，见下文。

discover()：统一发现引擎的内部机制

discover()是全部服务发现方法的引擎，签名与参数语义如下（mdns_discovery.py）：

该方法带有一组参数合法性校验，违反即抛ValueError：

• all_services与service_types不能同时指定；
• service_types不能是空列表；
• query_timeout_sec仅在query_service=True时允许非默认值；
• append_results仅在query_service=True时允许为True。

发现过程（源码级拆解）

discover(query_service=True) │ ▼ 确定要浏览的类型列表（all_services 时先扫描全部服务类型） │ ▼ async with AsyncZeroconf(interfaces=self.interfaces) as azc: ├── 创建 AsyncServiceBrowser(zeroconf, type_, handlers=[_on_service_state_change]) ├── 后台任务 _monitor_discovery_silence()：连续 2 秒无新服务即提前结束 ├── await wait_for(self._event.wait(), timeout=discovery_timeout_sec) │ └── 每个 Added 状态的服务触发 _on_service_state_change()，收集 PTR 记录 └── 若 query_service=True： ├── 用 Semaphore(5) 限制最多 5 个并发查询 ├── 对每条 PTR 调用 _query_service_info()（SRV/TXT/A/AAAA 全查） └── 结果按服务类型存入 self._discovered_services ▼ 返回 list[MdnsServiceInfo]（或 dict 映射）

几个值得注意的工程细节：

• 静默提前结束：_monitor_discovery_silence()（mdns_discovery.py）每 0.5 秒轮询一次，只要已经发现过至少一个服务、且超过DISCOVERY_SILENCE_THRESHOLD_SEC（2 秒）没有新服务出现，就提前置位事件结束 browse。TC_SC_4_1 等测试用例的注释也印证了这一点："Browses that get an answer end early via MdnsDiscovery's discovery-silence"。这使发现过程在典型局域网上远快于 15 秒上限；
• 并发保护：记录查询阶段通过Semaphore(5)将并发 mDNS 查询限制为 5 个，避免在设备较多时对系统造成过载；
• 去重：_on_service_state_change()只响应ServiceStateChange.Added，并以 service name 集合去重，防止重复 PTR 记录。

记录查询内部流程

_query_service_info()（mdns_discovery.py）是所有记录查询的核心私有方法，流程如下：

_query_service_info(service_type, service_name, query_record_types) │ ▼ async with AsyncZeroconf(...) as azc: ├── 注册 MdnsServiceListener，等待服务 add/update 事件 │ （wait_for_service_update，超时 SERVICE_LISTENER_TIMEOUT_SEC=5 秒） ├── 构造 MdnsAsyncServiceInfo，设置要查询的记录类型集合 ├── service_info.async_request(zc, timeout_ms=query_timeout_sec*1000) └── 移除监听器；成功则包装为 MdnsServiceInfo 返回 ▼ returns MdnsServiceInfo | None

其中的MdnsAsyncServiceInfo.async_request()重写自zeroconf.ServiceInfo，与基类实现的关键差异（见 mdns_async_service_info.py）：

• 绕过 known-answer 缓存：每次调用先清空zc.cache、zc.question_history与自身缓存，强制发起一次全新的网络查询，保证拿到新鲜数据；
• QU 优先 + QM 兜底：首次发送以单播优先（as_qu=True）的查询，随后始终发送组播（QM）查询，确保覆盖所有应答者；
• 防重复问题抑制：后续重试间隔取_DUPLICATE_QUESTION_INTERVAL与初始延迟 + 随机偏移中的较大者，避免因同一问题重发过快被设备抑制应答；
• 完成后的滞留监听：记录齐备后再等待 300–500 ms 随机时长，被动捕获迟到的 SRV/TXT/AAAA/A 应答；
• 循环以_is_complete与绝对截止时间（now_ms + timeout_ms）双条件控制，超时返回False。

各记录查询方法流程示意

get_srv_record/get_txt_record（以get_srv_record为例）：

get_srv_record(service_name, service_type) │ ▼ _query_service_info(..., query_record_types={SRV}) ├── 添加服务监听器（add_service_listener） ├── async_request(...) 发起 SRV 查询 └── 返回 MdnsServiceInfo 对象 ▼ returns MdnsServiceInfo | None

get_quada_records（AAAA 解析）：

get_quada_records(hostname) │ ▼ addr_resolver = AddressResolverIPv6(server=hostname) │ ├── addr_resolver.async_request(...) │ └── 发起 AAAA 记录的 mDNS 查询（_query_record_types={_TYPE_AAAA}） │ └── addr_resolver.ip_addresses_by_version(IPVersion.V6Only) └── 提取解析出的 IPv6 地址 ▼ returns list[AaaaRecord]

get_ptr_records（纯 PTR 浏览，不做完整解析）：

get_ptr_records(service_types) │ ▼ discover(query_service=False) │ └── 启动 AsyncServiceBrowser(...) └── 每个服务触发 _on_service_state_change(...) ├── 收集 PTR 记录信息 └── 存入 self._discovered_services ▼ returns list[PtrRecord]

get_all_service_types：

get_all_service_types() │ ▼ AsyncZeroconfServiceTypes.async_find(...) │ └── 扫描所有被公告的 mDNS 服务类型（超时返回空列表） ▼ returns List[str]

get_commissionable_subtypes：

get_commissionable_subtypes() │ ▼ get_all_service_types() │ └── 发现所有被公告的 mDNS 服务类型 ▼ 筛选包含 '._sub._matterc._udp.local.' 的 subtype ▼ returns List[str]

数据模型：MdnsServiceInfo、PtrRecord 与 AaaaRecord

MdnsDiscovery的三类返回容器定义在 data_classes 目录，均实现json_dict()序列化（基类见 json_serializable.py），配合log_output=True时可输出结构化的 JSON 发现报告。

MdnsServiceInfo

对zeroconf的AsyncServiceInfo的轻量包装（mdns_service_info.py），通过属性映射暴露常用字段：

PtrRecord

PTR 记录的容器（ptr_record.py），构造时自动从service_type与service_name推导instance_name：若服务类型含._sub.，取其后基类型（如_matter._tcp.local.）再从服务名中剥离，得到纯实例名（如A6666A3E45CF5655）。

AaaaRecord

IPv6 地址记录容器（aaaa_record.py），在__post_init__中完成大量地址分类元数据的生成：

• address：压缩格式 IPv6 地址（剥离去 zone index，如%eth0）；
• interface：将 scope id 映射为接口名（通过socket.if_nameindex()）；
• type_info：is_global/is_link_local/is_loopback/is_multicast分类标志；
• special_types：teredo/sixtofour/is_reserved/ipv4_mapped等特殊地址类型；
• meta：max_prefixlen与 packed 十六进制字节。

发现结果示例

以下为 README 与源码共同展示的典型发现结果样例，有助于理解各字段含义。

AsyncZeroconfServiceTypes 返回的服务类型示例：

_matterd._udp.local. _matter._tcp.local. _V65521._sub._matterd._udp.local. _IM7322C948581262F._sub._matter._tcp.local.

（_IM...为 operational 节点公告的实例名 subtype，_V65521...为 vendor subtype。）

AsyncServiceBrowser 返回的 PTR 记录信息示例：

"service_type": "_matterd._udp.local." "service_name": "A6666A3E45CF5655._matterd._udp.local." "instance_name": "A6666A3E45CF5655"

async_request 解析出的完整服务信息示例：

"service_name": "354D34458F15657D._matterd._udp.local." "service_type": "_V65521._sub._matterd._udp.local." "instance_name": "354D34458F15657D" "hostname": "00155DD54A04.local." "port": 5550 "addresses": ["172.30.139.182", "fe80::215:5dev:fed5:4a04"] "txt": {"VP": "65521+32769"} "priority": 0 "interface_index": 2 "weight": 0 "host_ttl": 120 "other_ttl": 4500

mDNS 值校验断言（utils/asserts.py）

asserts.py 提供了一套面向 mDNS 发现测试的校验助手。所有断言基于Mobly框架（from mobly import asserts），失败时抛出TestFailure，并且错误消息会明确指出违反的具体约束，便于测试快速定位失败原因。所有函数都带@not_none_args装饰器，保证参数不为None。

典型用法：

from mdns_discovery.utils.asserts import assert_valid_dn_key assert_valid_dn_key("Kitchen")

断言函数清单

其中两个断言与测试中的 TCP 能力判定关系密切：

• assert_txt_record_present(instance_name, service_type)先构造{instance_name}.{service_type.value}完整 QNAME，调用MdnsDiscovery().get_txt_record()获取记录，然后校验记录非空且含键值对；
• assert_valid_t_key用于将 DUT 实际公告的T键与 PICS（Protocol Implementation Conformance Statement）中声明的 TCP 支持对齐：TCP 支持时T ∈ {4, 6}（IPv4/IPv6），仅 MRP 时T = 0；此外 DUT 与 PICS 的 TCP 支持必须一致，否则直接断言失败。utils/network.py中的is_dut_tcp_supported()与之配套：T 键缺失、为空或为"0"视为仅支持 MRP（不支持 TCP），其他值视为支持 TCP。

单元测试见 tests/test_asserts.py，采用unittest框架对每个断言的正反用例（正确值通过、错误类型/空串/缺尾点/边界值失败）做了系统覆盖。例如TestAssertIsBorderRouterType验证了正确值通过、错误服务类型失败、空串失败、缺少尾随点失败等场景，并检查失败消息包含期望值。

网络工具函数（utils/network.py）

network.py 提供两个支持 mDNS 测试的网络工具：

get_host_ipv6_addresses()

扫描本机所有网络适配器的 IPv6 地址并过滤不可用项（network.py）：

• 跳过回环地址::1；
• link-local 地址（fe80::/10）必须附带接口作用域%接口名（如fe80::xxxx%eth0）才能用于通信；
• ULA（fd00::/8）与全局地址（2000::/3）可直接使用，保留纯地址；
• 若完全找不到可用 IPv6 地址，则返回InterfaceChoice.All，交由 Zeroconf 使用默认接口选择策略。

该函数返回的地址列表正是MdnsDiscovery.__init__中self.interfaces的来源，确保 mDNS 查询绑定到正确的 IPv6 接口。

使用示例：

from mdns_discovery.utils.network import get_host_ipv6_addresses addr_list = get_host_ipv6_addresses() for addr in addr_list: print(addr)

is_dut_tcp_supported()

异步函数，通过查询 DUT 的 operational 服务 TXT 记录判断其是否支持 TCP（network.py）：

• 构造MdnsDiscovery()并调用get_txt_record(service_name=instance_qname, service_type=OPERATIONAL.value)；
• 读取 TXT 记录中的T键：缺失、空串或"0"→ 返回False（仅 MRP 支持，不支持 TCP）；
• 其他值 → 返回True。

在 Matter 测试框架中的真实应用

MdnsDiscovery不是孤立的演示代码，而是 Matter Python 测试体系中的实际组件。以下是仓库内的真实调用场景：

1. DNS-SD 状态预检（commissioning.py）

_is_device_operational_via_dnssd()在建立 CASE 会话前先做 DNS-SD 预检：

compressed_fabric_id = dev_ctrl.GetCompressedFabricId() expected_instance_name = f'{compressed_fabric_id:016X}-{node_id:016X}' mdns = MdnsDiscovery() services = await mdns.get_operational_services( discovery_timeout_sec=discovery_timeout_sec, log_output=False ) for service in services: if service.instance_name == expected_instance_name: return True

其核心思想是：从设备控制器取得 compressed fabric ID 与 node ID，构造期望的 operational 实例名（{fabric:016X}-{node:016X}），再通过get_operational_services()遍历已发现服务匹配。匹配成功即认为设备已在该 fabric 上处于 operational 状态，从而避免不必要的 CASE 超时。同理，_is_device_commissionable_via_dnssd()通过get_commissionable_services()检测设备是否正处于配网模式。

2. Discriminator Subtype 验证（TC_SC_4_1.py）

TC_SC_4_1 测试用例构造 Discriminator Subtype：

discriminator_subtype = f"_L{discriminator}._sub.{MdnsServiceType.COMMISSIONABLE.value}" assert_valid_long_discriminator_subtype(discriminator_subtype)

随后调用：

ptr_records = await MdnsDiscovery().get_ptr_records( service_types=[discriminator_subtype], log_output=True ) asserts.assert_equal(len(ptr_records), 1, ...)

验证开放配网窗口期间，网络上恰好公告一条与判别器匹配的 subtype PTR 记录。这展示了「断言构造 → PTR 查询 → 结果校验」的完整测试闭环。

3. 其他测试用例引用

mdns_discovery还被 TC_CADMIN_1_15、TC_CNET_4_12、TC_DD_3_24、TC_ICDM_5_1、TC_JFADMIN_2_2、TC_SC_4_3、TC_SC_4_6、TC_SC_4_7 以及 commissioning.py 等模块引用，覆盖管理员、网络、DD、ICD、联合 Fabric、SC 等多个测试域的 mDNS 验证需求。

环境依赖与使用前提

• 该模块位于 src/python_testing/mdns_discovery，导入路径为mdns_discovery.*，运行环境需将src/python_testing加入PYTHONPATH；
• 依赖python-zeroconf（zeroconf/zeroconf.asyncio）与ifaddr（接口枚举）两个第三方库；
• 断言模块依赖Mobly测试框架（mobly.asserts、mobly.signals.TestFailure）；
• 发现过程基于IPv6：运行主机必须配置至少一个可用的 IPv6 接口地址（链路本地或路由地址均可），否则get_host_ipv6_addresses()会退回 Zeroconf 默认接口选择；
• 所有方法均为异步（async def），需要在asyncio事件循环中调用（可用asyncio.run()）。

总结与进一步阅读

MdnsDiscovery为 Matter 生态提供了一套完整、可控、可验证的 mDNS 服务发现方案：高层方法直接对应 Matter 规范的四种角色服务类型；低层记录查询暴露 PTR/SRV/TXT/AAAA 原始记录；discover()引擎内置静默提前结束、并发限制与缓存绕过机制，兼顾效率与数据新鲜度；配套断言库则把 Matter 规范中对实例名、subtype、TXT 键值的格式约束固化为可复用的测试校验器。

若要深入了解每个方法的完整参数说明，可直接阅读 mdns_discovery.py 中的内联 docstring；Matter 规范中对实例名、hostname 与 TXT 键的构造要求，可对照 asserts.py 中各断言的 docstring 与约束实现；完整的断言正反用例可查看 test_asserts.py。

【免费下载链接】connectedhomeipMatter (formerly Project CHIP) creates more connections between more objects, simplifying development for manufacturers and increasing compatibility for consumers, guided by the Connectivity Standards Alliance.项目地址: https://gitcode.com/GitHub_Trending/co/connectedhomeip