局域网服务发现终极指南:掌握 Python ZeroConf,让你的应用自动互相“找到彼此”

pip install zeroconf

# 同步 API（核心）from zeroconf import (    Zeroconf,              # 主入口类    ServiceInfo,           # 服务信息描述    ServiceBrowser,        # 服务浏览器    ServiceListener,       # 监听器基类    ServiceStateChange,    # 服务状态变更枚举    ZeroconfServiceTypes,  # 浏览当前网络的所有服务类型    IPVersion,             # IPv4 / IPv6 选择)# 异步 API（asyncio 集成）from zeroconf.asyncio import (    AsyncZeroconf,    AsyncServiceBrowser,    AsyncServiceInfo,)

from zeroconf import Zeroconf, ServiceInfoimport socket# ============================================================# 第一步：获取本机 IP 地址# ============================================================def get_local_ip() -> str:    """获取本机局域网 IP 地址"""    s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)    try:        # 不需要真的连接，只是让系统选择出站接口        s.connect(("8.8.8.8", 80))        return s.getsockname()[0]    finally:        s.close()# ============================================================# 第二步：构造服务信息# ============================================================local_ip = get_local_ip()service_type = "_http._tcp.local."       # HTTP 服务类型service_name = f"MyWebServer.{service_type}"  # 完整服务名info = ServiceInfo(    type_=service_type,          # 服务类型，必须是 "_<协议>._<传输>.local." 格式    name=service_name,           # 完整服务实例名    addresses=[socket.inet_aton(local_ip)],  # IP 地址列表（二进制格式）    port=8080,                   # 服务端口    properties={                 # TXT 记录（键值对，key 和 value 必须是 bytes）        b"version": b"1.0.0",        b"author": b"YourName",        b"path": b"/api",    },    server=f"MyHost.local.",     # 服务器主机名)# ============================================================# 第三步：注册服务# ============================================================zc = Zeroconf()try:    zc.register_service(info)    print(f"✅ 服务已注册: {service_name} @ {local_ip}:8080")    print("   按 Ctrl+C 退出...")    input()  # 保持运行，等待用户中断finally:    zc.unregister_service(info)    zc.close()    print("🛑 服务已注销")

MyWebServer._http._tcp.local.  → 地址: 192.168.1.100  → 端口: 8080  → TXT: version=1.0.0, author=YourName, path=/api

from zeroconf import Zeroconf, ServiceBrowser, ServiceListenerfrom typing import castimport socketclass MyServiceListener(ServiceListener):    """    服务发现监听器。    三个核心回调：add_service、remove_service、update_service。    """    def __init__(self):        self.discovered: dict[str, dict] = {}  # 用 name 索引已发现服务    def add_service(self, zc: Zeroconf, type_: str, name: str) -> None:        """新服务被发现"""        info = zc.get_service_info(type_, name)        if info is None:            return        # 解析 IP 地址（二进制 → 可读字符串）        addresses = [            socket.inet_ntoa(cast(bytes, addr))            for addr in info.addresses        ]        service_data = {            "name": info.name,            "type": info.type,            "server": info.server,            "addresses": addresses,            "port": info.port,            "txt": {                k.decode("utf-8"): v.decode("utf-8")                for k, v in (info.properties or {}).items()            },        }        self.discovered[name] = service_data        print(f"🟢 发现服务: {name}")        print(f"   地址: {addresses}")        print(f"   端口: {info.port}")        print(f"   TXT: {service_data['txt']}")    def remove_service(self, zc: Zeroconf, type_: str, name: str) -> None:        """服务离线"""        self.discovered.pop(name, None)        print(f"🔴 服务离线: {name}")    def update_service(self, zc: Zeroconf, type_: str, name: str) -> None:        """服务信息变更（端口、IP 或 TXT 更新）"""        print(f"🔄 服务更新: {name}")        # 重新拉取最新信息        self.add_service(zc, type_, name)# ============================================================# 启动发现# ============================================================zc = Zeroconf()listener = MyServiceListener()browser = ServiceBrowser(zc, "_http._tcp.local.", listener)print("🔍 开始扫描局域网 _http._tcp.local. 服务...")print("   按 Enter 退出\n")try:    input()  # 阻塞等待finally:    browser.cancel()    zc.close()    print("\n🛑 浏览器已关闭")    print(f"共发现 {len(listener.discovered)} 个服务:")    for name, data in listener.discovered.items():        print(f"   · {name} → {data['addresses'][0]}:{data['port']}")

🔍 开始扫描局域网 _http._tcp.local. 服务...   按 Enter 退出🟢 发现服务: MyWebServer._http._tcp.local.   地址: ['192.168.1.100']   端口: 8080   TXT: {'version': '1.0.0', 'author': 'YourName', 'path': '/api'}🟢 发现服务: RaspberryPi._http._tcp.local.   地址: ['192.168.1.200']   端口: 80   TXT: {'platform': 'raspberry-pi'}

from zeroconf import ZeroconfServiceTypesprint("📡 正在扫描局域网中所有 mDNS 服务类型...\n")all_types = ZeroconfServiceTypes.find()for i, stype in enumerate(all_types, 1):    print(f"   {i:>3}. {stype}")print(f"\n共发现 {len(all_types)} 种服务类型")

   1. _adb-tls-connect._tcp.local.   2. _airplay._tcp.local.   3. _esphomelib._tcp.local.   4. _hap._tcp.local.                ← HomeKit   5. _http._tcp.local.   6. _printer._tcp.local.   7. _ssh._tcp.local.   8. _workstation._tcp.local.

import asynciofrom zeroconf import Zeroconf, ServiceStateChangefrom zeroconf.asyncio import AsyncZeroconf, AsyncServiceBrowser, AsyncServiceInfoasync def async_browser_callback(    zeroconf: Zeroconf,    service_type: str,    name: str,    state_change: ServiceStateChange,) -> None:    """异步回调 —— 在事件循环中直接处理服务事件"""    if state_change == ServiceStateChange.Added:        print(f"🟢 [async] 发现: {name}")        # 异步请求完整服务信息，不阻塞事件循环        info = AsyncServiceInfo(service_type, name)        if await info.async_request(zeroconf, timeout=3000):            addrs = [f"{a}" for a in info.parsed_addresses()]            print(f"   地址: {addrs}, 端口: {info.port}")    elif state_change == ServiceStateChange.Removed:        print(f"🔴 [async] 离线: {name}")async def main():    """异步入口"""    aio_zc = AsyncZeroconf()    # 创建异步浏览器    browser = AsyncServiceBrowser(        aio_zc.zeroconf,        "_http._tcp.local.",        handlers=[async_browser_callback],  # 支持多个回调    )    print("🔍 [async] 开始扫描（10 秒后自动停止）...")    await asyncio.sleep(10)    await browser.async_cancel()    await aio_zc.async_close()    print("🛑 [async] 扫描结束")if __name__ == "__main__":    asyncio.run(main())

from zeroconf import Zeroconf, ServiceBrowser, ServiceListenerclass MultiTypeListener(ServiceListener):    def add_service(self, zc, type_, name):        print(f"🟢 [{type_}] 新服务: {name}")    def remove_service(self, zc, type_, name):        print(f"🔴 [{type_}] 离线: {name}")    def update_service(self, zc, type_, name):        print(f"🔄 [{type_}] 更新: {name}")zc = Zeroconf()listener = MultiTypeListener()# 同时监控 HTTP、SSH 和 ESPHome 设备types_to_monitor = [    "_http._tcp.local.",    "_ssh._tcp.local.",    "_esphomelib._tcp.local.",   # ESPHome 固件设备的专属类型]browsers = []for stype in types_to_monitor:    browser = ServiceBrowser(zc, stype, listener)    browsers.append(browser)try:    input("按 Enter 停止监控...\n")finally:    for b in browsers:        b.cancel()    zc.close()