Python UDP局域网通信原理与丢包分析实战

1. 标题里的“网络攻击”四个字，到底在说什么？

看到标题里“Python使用socket进行局域网内UDP协议的通信与网络攻击”，很多人第一反应是：这不就是教人写DDoS脚本？或者搞端口扫描？甚至联想到渗透测试、红队演练——但我要先说清楚：这个标题本身存在严重歧义，而这种歧义恰恰是绝大多数初学者踩坑的第一步。

我带过不少刚接触网络编程的开发者，他们拿着类似标题的教程，兴致勃勃写完几行socket.sendto()，发现发出去的数据对方收不到；再一查资料，又看到“UDP攻击”“UDP Flood”这类词，立刻陷入困惑：为什么同一个协议，一边是“通信”，一边是“攻击”？它到底是工具，还是武器？

答案很朴素：UDP本身没有善恶，它只是一条没有门卫、不登记访客、也不确认收货的快递通道。你用它寄一份购物清单，是通信；你用它往同一地址狂轰滥炸一万份相同清单，让收件人根本来不及拆包、直接堆满仓库瘫痪服务——这就构成了“攻击”的技术基础。关键不在协议，而在发送意图、数据量级、目标承受能力、以及是否获得授权。

所以，本文要做的第一件事，不是教你“怎么攻击”，而是帮你把这句话刻进本能：所有局域网内的UDP通信实验，必须严格限定在你完全控制的封闭环境里——比如一台物理机+一个VirtualBox虚拟机，或两台连在同一根网线、未接入任何外部网络的树莓派。我亲眼见过有开发者在公司内网跑UDP发包脚本，结果触发了IT部门的流量异常告警系统，被叫去写了三页《关于UDP广播行为的说明》。这不是危言耸听，是真实发生过的教训。

关键词里虽然空着，但标题已明确锚定三个不可绕开的技术坐标：Python、socket、UDP、局域网。这意味着我们不谈TCP三次握手、不聊TLS加密、不碰公网IP和NAT穿透——所有内容都压在最底层、最原始、也最容易暴露问题的通信层上。这种“裸奔式”通信，恰恰是最能锤炼网络直觉的训练场。

接下来的内容，我会带你从零搭起一对可验证的UDP收发端，然后一层层剥开那些藏在sendto()和recvfrom()背后的真实约束：为什么你发1000次，对方只收到982次？为什么加个time.sleep(0.001)反而更不稳定？为什么用localhost能通，换成本机真实IP就收不到？这些都不是Bug，而是UDP协议在真实硬件和操作系统调度下必然呈现的“本来面目”。

提示：本文所有代码均基于Python 3.8+标准库，不依赖任何第三方包。你不需要懂Wireshark，但需要一台能打开两个终端窗口的电脑——这就是全部硬件要求。

2. 从“Hello World”到“Hello, I’m Here”：UDP通信的最小可行闭环

很多教程一上来就贴一大段服务端+客户端代码，然后说“运行就能通”。结果读者照着敲，服务端没报错，客户端也没报错，但就是没收到消息。问题出在哪？往往卡在最基础的地址绑定逻辑上。

我们先写一个真正能跑通、且每一步都可验证的最小闭环。不是为了炫技，而是为了建立确定性——当你知道哪一行代码对应哪个网络动作时，后续排查才有依据。

2.1 服务端：不叫“Server”，叫“Listener”

# listener.py import socket import sys def start_listener(host='0.0.0.0', port=8888): # 创建UDP socket：SOCK_DGRAM代表数据报（即UDP），AF_INET代表IPv4 sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) # 关键一步：绑定地址和端口 # host='0.0.0.0' 表示监听本机所有IPv4地址（包括127.0.0.1和局域网IP） # 如果写成 '127.0.0.1'，则只能接收来自本机的包，收不到其他设备发来的 try: sock.bind((host, port)) print(f"✅ Listener started on {host}:{port}") print(" Waiting for UDP packets...") except OSError as e: print(f"❌ Bind failed: {e}") print(" Hint: Port might be occupied, or you're using wrong host address.") sys.exit(1) # 持续接收，直到手动中断（Ctrl+C） while True: try: # recvfrom() 是阻塞式调用：没有数据来，就一直等 # 1024 是缓冲区大小（字节），不是单次最大接收长度，而是socket内部一次读取的上限 data, addr = sock.recvfrom(1024) message = data.decode('utf-8').strip() print(f"📩 Received from {addr[0]}:{addr[1]} -> '{message}'") # 可选：回一个确认包（体现双向通信） ack = f"ACK: {len(message)} chars" sock.sendto(ack.encode('utf-8'), addr) except KeyboardInterrupt: print("\n👋 Listener stopped.") break except UnicodeDecodeError: # UDP可能收到非UTF-8编码的二进制数据，这里做容错 print(f"⚠️ Received non-UTF8 data from {addr}, length: {len(data)} bytes") except Exception as e: print(f"❗ Unexpected error: {e}") if __name__ == '__main__': start_listener()

这段代码里，有三个极易被忽略但决定成败的细节：

1. bind()的host参数不是可有可无的装饰
   很多人习惯性写bind(('localhost', 8888))或bind(('127.0.0.1', 8888))。这在本机测试时看似能通，但一旦换成局域网另一台设备发包，就彻底失联。因为127.0.0.1是环回地址，只响应本机进程间的通信，不会接收来自物理网卡的任何数据包。正确做法是'0.0.0.0'（监听所有IPv4接口）或明确指定本机局域网IP（如'192.168.1.100'）。

2. recvfrom()的缓冲区大小1024，并非“安全值”
   UDP单个数据报理论最大65507字节（64KB减去IP头20B+UDP头8B），但实际中，超过1500字节就大概率触发IP分片。而分片后的UDP包，任意一片丢失，整个数据报就作废。所以工业级应用通常将单次UDP载荷控制在1400字节以内。我们这里用1024，是为兼顾教学清晰度与实操鲁棒性——足够传文本，又避开分片陷阱。

3. decode('utf-8')前的.strip()不是优雅，而是必要
   UDP不保证数据边界。你发b"hello"，对方可能收到b"hello\n\0\0\0"（末尾填充了空字节）。.strip()清掉首尾空白，避免打印出乱码或误判为空消息。这是处理原始二进制协议时的“肌肉记忆”。

2.2 客户端：不叫“Client”，叫“Sender”

# sender.py import socket import sys import time def send_message(host, port, message, count=1, interval=0.1): sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) # UDP客户端通常不需要bind()，系统会自动分配临时端口 # 但如果你想固定源端口（比如用于防火墙策略），可以显式bind # sock.bind(('0.0.0.0', 55555)) # 示例：固定源端口55555 print(f"📤 Sending {count} messages to {host}:{port}") for i in range(count): try: # sendto() 是无连接的：不检查对方是否存在，不等待确认 sock.sendto(message.encode('utf-8'), (host, port)) print(f" [{i+1}] Sent: '{message}'") # 短暂休眠，避免瞬间发包过多导致本地缓冲区溢出 if i < count - 1: # 最后一次不sleep，保持节奏感 time.sleep(interval) except OSError as e: print(f"❌ Send failed on attempt {i+1}: {e}") print(" Hint: Check if target host/port is reachable and listening.") break except Exception as e: print(f"❗ Unexpected error on attempt {i+1}: {e}") break sock.close() print("👋 Sender finished.") if __name__ == '__main__': if len(sys.argv) < 4: print("Usage: python sender.py <host> <port> <message> [count] [interval]") print("Example: python sender.py 192.168.1.100 8888 'Hello UDP'") sys.exit(1) host = sys.argv[1] port = int(sys.argv[2]) message = sys.argv[3] count = int(sys.argv[4]) if len(sys.argv) > 4 else 1 interval = float(sys.argv[5]) if len(sys.argv) > 5 else 0.1 send_message(host, port, message, count, interval)

这个客户端的设计哲学是：暴露所有可控变量，拒绝黑盒。

• host和port必须由用户显式输入，强迫你思考“我要发给谁？”
• count和interval默认为1和0.1秒，但允许你一键改成发1000次、间隔1毫秒——这正是理解“通信”与“攻击”边界的实验入口。

注意：运行前务必确认listener.py已在目标机器上启动，且防火墙放行了对应端口。Windows默认防火墙会拦截UDP入站，Linux的ufw或iptables也可能拦截。这是90%的“明明代码没错却收不到”的根源。别急着改代码，先查防火墙。

3. 当“发出去”不等于“收得到”：UDP丢包的七种真实原因与验证法

UDP的“不可靠”不是一句空话。它意味着：你调用sendto()成功返回，只代表数据已交给操作系统内核发送队列，并不保证抵达对端。在局域网这种“理想环境”下，丢包率仍可能高达1%~5%，远高于TCP的万分之一。下面这七种丢包场景，我都用真实抓包和日志复现过，每一种都有对应的验证方法。

3.1 场景一：本地发送队列溢出（Send Buffer Overflow）

现象：连续高速发包（如每毫秒1次），sendto()调用始终返回成功，但接收端收到数量远少于发送数，且无错误日志。

原理：Linux/Windows内核为每个UDP socket维护一个发送缓冲区（sndbuf）。当应用发包速度 > 内核将数据推送到网卡的速度时，缓冲区填满，后续sendto()会立即返回错误errno=ENOBUFS（No buffer space available）。但很多Python代码没检查返回值，误以为“发成功了”。

验证法：

1. 在sender.py的sendto()后加错误检查：

   try: sock.sendto(...) except OSError as e: if e.errno == errno.ENOBUFS: print("⚠️ Send buffer full! Dropping packet.") raise

2. 查看当前系统UDP发送缓冲区大小：

   # Linux cat /proc/sys/net/core/wmem_default # 默认值 cat /proc/sys/net/core/wmem_max # 最大值

3. 临时增大缓冲区（需root）：

   echo 4194304 > /proc/sys/net/core/wmem_max # 设为4MB

经验：在高吞吐场景（如实时音视频流），建议将sendto()包装成带重试和背压的函数，而非简单循环。

3.2 场景二：接收端缓冲区溢出（Recv Buffer Overflow）

现象：发送端稳定发包，接收端recvfrom()调用频率不够，导致部分包永远丢失，且无提示。

原理：recvfrom()每次只取缓冲区队首的一个UDP数据报。如果应用处理速度慢（如每次收包后做耗时计算），新包不断涌入，缓冲区（rcvbuf）满了，内核就会丢弃新到的包。这个过程静默发生，recvfrom()不会报错，只是“收不到”。

验证法：

1. 在listener.py中，recvfrom()前加时间戳，记录两次调用间隔：

   import time last_recv = time.time() while True: data, addr = sock.recvfrom(1024) now = time.time() gap = now - last_recv if gap > 0.1: # 超过100ms，说明处理变慢 print(f"⏱️ Long processing gap: {gap:.3f}s") last_recv = now # ... 处理data

2. 查看接收缓冲区大小：

   cat /proc/sys/net/core/rmem_default cat /proc/sys/net/core/rmem_max

解决方案：

• 增大接收缓冲区（同发送端）
• 用多线程：主线程只负责recvfrom()并存入队列，工作线程异步处理
• 使用select()或epoll()实现单线程高并发接收（进阶）

3.3 场景三：网卡驱动或硬件丢包（Driver/Hardware Drop）

现象：ping和TCP连接都正常，唯独UDP大量丢包；更换不同品牌网卡后问题消失。

原理：廉价USB网卡或老旧集成网卡的驱动，在高UDP流量下可能因DMA缓冲区管理缺陷而丢包。这不是协议问题，是硬件栈的锅。

验证法：

1. 查看网卡驱动统计：

   # Linux，查看eth0（替换为你网卡名） ethtool -S eth0 | grep -i "drop\|error\|over" # 关注 rx_dropped, rx_over_errors, rx_fifo_errors

2. 对比测试：用同一台机器，分别插USB网卡和主板自带网卡，跑相同UDP压力测试。

经验：企业级部署务必选用Intel或Broadcom芯片网卡，消费级Realtek RTL8153 USB网卡在UDP高负载下丢包率可达10%以上。

3.4 场景四：交换机端口缓存溢出（Switch Buffer Exhaustion）

现象：两台设备直连正常，接入千兆交换机后丢包率陡增；更换为万兆交换机后恢复。

原理：交换机每个端口有有限的包缓冲区（Buffer）。当某端口接收速率 > 转发到其他端口的速率时，缓冲区满，新包被丢弃。UDP无拥塞控制，会持续猛灌，加剧此问题。

验证法：

1. 登录交换机管理界面，查看端口统计中的Input Discards或Drops计数。
2. 用iperf3 -u -c <target> -b 100M模拟UDP流，观察丢包率变化。

对策：

• 避免在核心交换机上做UDP洪泛测试
• 启用QoS策略，为UDP流限速（如限制单流不超过50Mbps）
• 物理层优化：确保网线为Cat5e及以上，避免使用劣质HUB替代交换机

3.5 场景五：操作系统防火墙拦截（OS Firewall Block）

现象：listener.py进程在运行，netstat -anu | grep :8888显示端口已监听，但sender.py发包后listener无任何输出。

原理：Windows Defender Firewall、Linux的ufw或iptables默认阻止UDP入站连接。netstat显示监听，只代表socket已创建，不代表数据能进来。

验证法：

1. 临时关闭防火墙测试（仅限实验室环境！）：

   # Windows (管理员PowerShell) Set-NetFirewallProfile -Profile Domain,Private,Public -Enabled False # Linux (Ubuntu) sudo ufw disable

2. 若关闭后通信恢复，则确认是防火墙问题。

永久方案：

• Windows：在防火墙高级设置中，新建入站规则，协议UDP，端口8888，允许连接
• Linux：sudo ufw allow 8888/udp或sudo iptables -A INPUT -p udp --dport 8888 -j ACCEPT

3.6 场景六：ARP缓存未命中（ARP Cache Miss）

现象：首次发包延迟极高（>1秒），后续正常；重启网络后重现。

原理：UDP发包前，需通过ARP协议获取目标IP对应的MAC地址。若本地ARP缓存中没有该条目，系统会发ARP请求并等待响应（默认超时1秒），期间sendto()阻塞。

验证法：

1. 发包前清空ARP缓存：

   # Linux ip neigh flush all # Windows arp -d *

2. 用tcpdump抓包，观察是否有ARP Request发出。

解决：

• 首次通信前，主动ping目标IP一次，触发ARP学习
• 在/etc/hosts或WindowsC:\Windows\System32\drivers\etc\hosts中静态绑定IP-MAC（适用于固定设备）

3.7 场景七：UDP校验和错误（Checksum Mismatch）

现象：listener.py收到数据，但data内容是乱码或长度异常；Wireshark显示该包标记为Bad checksum。

原理：UDP头部含16位校验和，用于检测传输错误。若网卡开启UDP checksum offload，校验和由硬件计算，但某些虚拟化环境（如VirtualBox）或驱动bug会导致计算错误，内核收到后因校验失败直接丢弃。

验证法：

1. 在Wireshark过滤器中输入udp && udp.checksum_bad == 1
2. 查看网卡offload状态：

   # Linux ethtool -k eth0 | grep checksum

修复：

• 关闭网卡UDP校验和卸载（需root）：

  ethtool -K eth0 tx off rx off

• 或更新网卡驱动至最新版

总结：UDP丢包不是玄学。每一次丢包，背后都有可定位的软硬件环节。养成用ethtool、tcpdump、netstat三件套交叉验证的习惯，比盲目改代码高效十倍。

4. 从“发1000次”到“理解Flood”：UDP Flood的底层机制与防御视角

标题里的“网络攻击”，在专业语境中特指UDP Flood——一种利用UDP无连接、无状态特性，向目标发送海量伪造源IP的UDP包，耗尽其带宽、连接表或处理资源的拒绝服务（DoS）手段。它不是黑客电影里的炫酷代码，而是一段只有十几行、但足以让普通服务器瘫痪的脚本。

但请注意：在未获明确书面授权的情况下，对任何非自己完全控制的设备执行UDP Flood，均属违法行为。本文仅解析其技术原理，目的是让你理解防御逻辑，而非提供攻击工具。

4.1 最简UDP Flood脚本：解剖它的每一行

# udp_flood_demo.py （仅用于本地环回测试！） import socket import random import threading import time def flood_target(host, port, duration=10): """向目标发送UDP洪水，仅用于理解原理""" sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) # 构造随机数据包（模拟不同应用层协议） payloads = [ b'\x00\x01\x00\x00\x00\x01\x00\x00\x00\x00\x00\x00\x07example\x03com\x00\x00\x01\x00\x01', # DNS查询 b'\x16\x03\x01\x00\xdc\x01\x00\x00\xd8\x03\x03', # TLS Client Hello片段 b'GET / HTTP/1.1\r\nHost: example.com\r\n\r\n', # HTTP片段 ] start_time = time.time() sent_count = 0 print(f"💥 Starting UDP flood to {host}:{port} for {duration}s...") while time.time() - start_time < duration: try: # 随机选择payload和随机端口（模拟反射放大） payload = random.choice(payloads) # 伪造源端口（实际中还会伪造源IP，需raw socket权限） fake_port = random.randint(1024, 65535) sock.sendto(payload, (host, port)) sent_count += 1 # 控制发包速率：此处设为1000包/秒 time.sleep(0.001) except Exception as e: print(f"⚠️ Send error: {e}") break sock.close() print(f"🏁 Flood ended. Total sent: {sent_count}") # 本地测试：只发给自己，不对外网产生影响 if __name__ == '__main__': # 重要：仅限此地址！ flood_target('127.0.0.1', 8888, duration=5)

这段代码的关键点在于：

• random.choice(payloads)：真实攻击者会构造针对特定服务的畸形包（如DNS、NTP、SNMP），触发服务端复杂解析，消耗更多CPU。我们用预置的二进制片段模拟。
• fake_port：虽未伪造源IP（那需要AF_PACKET权限），但随机端口增加了连接跟踪表（conntrack）的负担。
• time.sleep(0.001)：精确控制发包速率。1000包/秒对千兆网卡只是毛毛雨，但对嵌入式设备或老服务器已是洪峰。

4.2 为什么UDP Flood有效？——三重资源耗尽模型

UDP Flood的杀伤力，源于它同时冲击目标的三层资源：

实测案例：某高校实验室的树莓派4B（4GB RAM）运行dnsmasq作为DNS服务器。当UDP Flood以5000包/秒攻击其53端口时，conntrack表在3分钟内填满（默认65536条），随后所有新DNS查询超时。这不是服务崩溃，而是内核连接跟踪子系统过载。

4.3 防御实践：从主机到网络的四道防线

防御UDP Flood不是靠一个开关，而是纵深防御体系。以下是我在多个项目中验证有效的四层措施：

第一层：主机级防护（Host-Level）

• 禁用不必要的UDP服务：ss -uln查看所有监听UDP端口，关闭rpcbind、avahi-daemon等非必需服务。
• 配置conntrack参数（Linux）：

  # 增大连接跟踪表 echo 524288 > /proc/sys/net/netfilter/nf_conntrack_max # 缩短UDP流超时（默认180秒，改为30秒加速回收） echo 30 > /proc/sys/net/netfilter/nf_conntrack_udp_timeout_stream # 启用哈希桶优化（内核4.12+） echo 1 > /proc/sys/net/netfilter/nf_conntrack_hashsize

• 应用层限速：在listener.py中加入令牌桶算法，对同一源IP每秒最多处理10个包。

第二层：防火墙规则（Firewall Rules）

• Linux iptables限速：

  # 对8888端口，限制单IP每秒最多5个新连接（UDP五元组） iptables -A INPUT -p udp --dport 8888 -m state --state NEW -m limit --limit 5/sec --limit-burst 10 -j ACCEPT iptables -A INPUT -p udp --dport 8888 -j DROP

• Windows高级防火墙：创建入站规则，操作→限制→配置连接数限制为10/秒。

第三层：交换机/路由器ACL（Network ACL）

• 在接入层交换机上配置：

  interface GigabitEthernet0/1 ip access-group UDP_FLOOD_IN in ! ip access-list extended UDP_FLOOD_IN deny udp any any eq 8888 fragments ! 拒绝分片包（常见攻击特征） deny udp any any eq 8888 packet-length gt 1024 ! 拒绝超大包 permit udp any any eq 8888

第四层：云服务商防护（Cloud WAF/DDoS）

• 阿里云DDoS高防IP：可清洗Tbps级UDP Flood，支持自定义清洗策略。
• AWS Shield Advanced：自动检测UDP反射攻击，集成WAF规则。

经验之谈：小团队不必追求“全防”，优先保障核心服务端口（如DNS 53、NTP 123、SNMP 161）。对非关键端口，直接DROP比REJECT更安全——REJECT会触发ICMP响应，反而暴露服务存在。

5. 超越Hello World：构建一个可落地的局域网UDP心跳监控系统

学完丢包分析和Flood原理，是时候把知识焊接到真实需求上了。我曾为某智能硬件实验室设计过一套局域网设备心跳监控系统，它用UDP实现，核心诉求就三点：轻量（设备资源有限）、低延迟（秒级故障感知）、抗丢包（容忍20%丢包率）。这套方案至今还在稳定运行，下面我把关键设计拆解给你。

5.1 协议设计：为什么不用JSON，而用二进制TLV？

很多开发者第一反应是：“发个JSON字符串呗，{"type":"heartbeat","id":"dev001","ts":1712345678}”。但这是对UDP的误用。

• JSON体积大：上述字符串长62字节，而二进制TLV只需16字节（1B type + 1B id_len + 5B id + 8B ts + 1B crc）
• 解析开销高：嵌入式设备CPU弱，JSON解析耗时是二进制unpack的5倍以上
• 校验脆弱：JSON无内置校验，单字节错误可能导致整个解析失败

我们采用极简TLV（Type-Length-Value）格式：

Python打包示例：

import struct import zlib def build_heartbeat_packet(device_id: str, timestamp_ns: int) -> bytes: """构建二进制心跳包""" device_bytes = device_id.encode('ascii') id_len = len(device_bytes) # 打包：type(1) + id_len(1) + id(N) + ts(8) payload = struct.pack('!B B', 0x01, id_len) + device_bytes payload += struct.pack('!Q', timestamp_ns) # Q = unsigned long long (8 bytes) # 计算CRC8（简化版，生产环境用标准CRC8） crc = 0 for b in payload: crc ^= b for _ in range(8): if crc & 0x80: crc = (crc << 1) ^ 0x07 # polynomial x^8+x^2+x^1+1 else: crc <<= 1 crc &= 0xFF return payload + bytes([crc]) # 使用 pkt = build_heartbeat_packet("esp32-01", 1712345678901234567) print(f"Packet size: {len(pkt)} bytes") # 输出：18 bytes

5.2 心跳逻辑：如何用UDP实现“可靠”的不可靠协议？

UDP本身不可靠，但我们可以通过应用层逻辑逼近可靠性：

• 发送端（设备）：每5秒发1个心跳包，不重传，不等待ACK（节省功耗）
• 接收端（监控中心）：为每个设备维护一个“最后收到时间戳”。若15秒内无新包，则标记为离线
• 抗抖动：允许±2秒时钟漂移，避免因设备RTC不准误判离线

监控中心核心逻辑：

# monitor_center.py import socket import struct import time from collections import defaultdict class HeartbeatMonitor: def __init__(self, bind_host='0.0.0.0', bind_port=9999): self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) self.sock.bind((bind_host, bind_port)) self.devices = defaultdict(lambda: {'last_seen': 0, 'status': 'offline'}) self.timeout_sec = 15 def verify_crc(self, data: bytes) -> bool: """验证CRC8校验和""" if len(data) < 2: return False crc_received = data[-1] payload = data[:-1] crc_calc = 0 for b in payload: crc_calc ^= b for _ in range(8): if crc_calc & 0x80: crc_calc = (crc_calc << 1) ^ 0x07 else: crc_calc <<= 1 crc_calc &= 0xFF return crc_calc == crc_received def handle_packet(self, data: bytes, addr: tuple): if len(data) < 11: # 最小包长：1+1+N+8+1，N>=1 return if not self.verify_crc(data): return try: # 解析：type(1), id_len(1), id(N), ts(8) pkt_type = data[0] if pkt_type != 0x01: # 非心跳包 return id_len = data[1] if len(data) < 11 + id_len: # 至少11字节 + ID长度 return device_id = data[2:2+id_len].decode('ascii') timestamp_ns = struct.unpack('!Q', data[2+id_len:10+id_len])[0] # 更新设备状态 self.devices[device_id]['last_seen'] = timestamp_ns self.devices[device_id]['status'] = 'online' self.devices[device_id]['ip'] = addr[0] except (UnicodeDecodeError, struct.error, KeyError): return def check_offline_devices(self): """扫描离线设备""" now_ns = time.time_ns() offline_list = [] for dev_id, info in self.devices.items(): if info['status'] == 'online' and (now_ns - info['last_seen']) > (self.timeout_sec * 1_000_000_000): info['status'] = 'offline' offline_list.append(dev_id) if offline_list: print(f"🚨 Offline devices: {offline_list}") def run(self): print("📡 Heartbeat Monitor started...") while True: try: data, addr = self.sock.recvfrom(1024) self.handle_packet(data, addr) # 每秒检查一次离线状态 if hasattr(self, '_last_check') is False or time.time() - getattr(self, '_last_check', 0) >= 1.0: self.check_offline_devices() self._