24VDC 与 POE 双供电冗余设计：网口温湿度记录仪在供电切换时的数据连续性测试

原创

盛世宏博小可

发布于 2026-09-22 11:08:56

1280

24VDC 与 POE 双供电冗余设计：网口温湿度记录仪在供电切换时的数据连续性测试

前面我们把以太网温湿度传感器的硬件架构、分布式组网、协议转换网关、Docker 化部署、InfluxDB 降采样、CKafka 异步解耦、TCP 保活、净化恒湿一体机控制逻辑、环境数据一体化误区、四合一一体机接入腾讯云 IoT、从八防到十防的设计规范等全部讲透了。

这一篇聚焦一个现场最容易被忽略、出事最致命的问题——当 24VDC 和 PoE 双供电之间发生切换时，网口温湿度记录仪的数据到底会不会断、断了多久、断了之后能不能无缝续传。

先说一个残酷的现场事实：

标书上写着"支持 24VDC + PoE 双供电冗余"，厂家彩页上画着两条供电线并列，看起来很美。但没有任何一家厂家告诉你：PoE 供电从断开到恢复，交换机端口需要重新协商，设备内部 DC-DC 需要重新启动，TCP 连接会断，Modbus TCP 会话会丢，MQTT 连接要重连，本地缓存要补传——这一连串动作加起来，数据到底断了多少秒？没人测过。直到验收那天你拔掉一根电源线，曲线上出现了一个 30 秒的空洞。

物联网 #Modbus #TCP/IP #UDP #POE供电 #腾讯云 #Wireshark #Python #InfluxDB #以太网温湿度传感器 #网口温湿度变送器 #机房监控

一、双供电冗余的"理论"与"现实"

1. 硬件层面的冗余架构

典型双供电网口温湿度记录仪的供电架构：

┌─────────────────────────────────────────────────┐
│          温湿度记录仪（以太网）                   │
│                                                 │
│  ┌──────────┐         ┌──────────────────┐     │
│  │ 24VDC    │         │ PoE (802.3af/at) │     │
│  │ 端子输入 │         │ RJ45 线对 3,6,   │     │
│  │ (接线盒) │         │ 1,2,4,5         │     │
│  └────┬─────┘         └───────┬──────────┘     │
│       │                       │                │
│       ▼                       ▼                │
│  ┌────────────────────────────────────┐         │
│  │         ORing 二极管 / 理想二极管   │         │
│  │         (自动切换，无触点)          │         │
│  └──────────────┬─────────────────────┘         │
│                 ▼                               │
│  ┌──────────────────────────────┐               │
│  │   DC-DC 变换 + 超级电容缓冲   │               │
│  │   (部分高端型号才有)           │               │
│  └──────────────┬───────────────┘               │
│                 ▼                               │
│  ┌──────────────────────────────┐               │
│  │   MCU + 以太网 PHY + 传感器   │               │
│  └──────────────────────────────┘               │
└─────────────────────────────────────────────────┘

理论上：ORing 二极管实现"或"逻辑，哪路有电用哪路，切换时间纳秒级，数据不应该断。

现实中：

环节	切换时发生了什么	对数据连续性的影响
PoE 断开	交换机端口 down → 设备 PHY 失去链路信号	以太网链路中断
PoE 恢复	交换机端口 up → 设备 PHY 重新协商（MDI/MDIX、速率）	链路恢复需要 2–5 秒
24VDC 断开	ORing 切换到 PoE → 如果内部有超级电容，MCU 不掉电	可能无缝
24VDC 恢复	ORing 切回 → DC-DC 输出纹波可能短暂增大	MCU 可能复位
以太网链路中断	TCP 连接断开，MQTT 断开	需要重连 + 会话恢复
设备复位	部分低端型号在供电切换时 MCU 会复位	数据断采 + 重启耗时 15–30 秒

二、测试方案设计

1. 测试目标

测试项	测量指标	通过标准
24VDC → PoE 切换	数据中断时长、TCP 重连时间、MQTT 重连时间	中断 ≤ 5s，恢复后补传完整
PoE → 24VDC 切换	同上	同上
24VDC 反复通断（抖动）	设备是否反复重启、数据是否持续	不反复重启，数据连续
PoE 反复通断	同上	同上
双路同时断电	本地缓存是否保住最后 N 条数据	缓存不丢，恢复后补传
切换期间 Modbus TCP 请求	请求是否超时、是否有脏数据	超时请求标记 quality=bad

2. 测试拓扑

┌──────────────┐     ┌────────────────┐     ┌──────────────┐
│  可编程直流   │     │                │     │              │
│  电源 24VDC  │────▶│  温湿度记录仪   │────▶│  管理型交换机 │
│  (可远程开关) │     │  PoE + 24VDC   │     │  (端口镜像)  │
└──────────────┘     │                │     └──────┬───────┘
                     └────────────────┘            │
                              ▲                    │
                              │ PoE               │ 镜像端口
                              │                    │
                     ┌────────┴────────┐           │
                     │  PoE 交换机      │           │
                     │  (端口可远程     │           │
                     │   关闭/开启)     │           │
                     └─────────────────┘           │
                                                   ▼
                                          ┌────────────────┐
                                          │  测试主机        │
                                          │  · Wireshark   │
                                          │  · Python 采集  │
                                          │  · InfluxDB    │
                                          └────────────────┘

3. 测试工具

工具	用途
可编程直流电源（Keysight/固纬）	远程控制 24VDC 通断
PoE 管理型交换机（支持 CLI 关端口）	远程控制 PoE 通断
Wireshark（端口镜像）	抓包分析 TCP/MQTT 断开与重连时序
Python 采集脚本	以 1s 间隔持续轮询 Modbus TCP，记录每次请求的响应时间和质量
InfluxDB	存储测试结果，用于事后分析中断时长
Grafana	可视化数据连续性

三、Wireshark 抓包分析：切换时到底发生了什么

1. 正常状态下的通信

No.  Time      Source          Destination     Protocol  Info
1    0.000000  192.168.10.100  192.168.10.21  TCP       502 → 54321 [SYN]
2    0.000012  192.168.10.21   192.168.10.100 TCP       54321 → 502 [SYN, ACK]
3    0.000020  192.168.10.100  192.168.10.21  TCP       502 → 54321 [ACK]
4    1.000015  192.168.10.100  192.168.10.21  Modbus    Read Holding Registers
5    1.000089  192.168.10.21   192.168.10.100 Modbus    Response: 2 registers
6    2.000014  192.168.10.100  192.168.10.21  Modbus    Read Holding Registers
7    2.000091  192.168.10.21   192.168.10.100 Modbus    Response: 2 registers

2. PoE 断开瞬间

No.  Time      Source          Destination     Protocol  Info
...  (正常通信)
48   47.000015 192.168.10.100  192.168.10.21  Modbus    Read Holding Registers
49   47.000089 192.168.10.21   192.168.10.100 Modbus    Response: temp=22.4, rh=55.8
50   48.000015 192.168.10.100  192.168.10.21  Modbus    Read Holding Registers
51   48.002100 192.168.10.100  192.168.10.21  TCP       [Retransmission] 502 → 54321
52   48.004200 192.168.10.100  192.168.10.21  TCP       [Retransmission] 502 → 54321
53   48.006300 192.168.10.100  192.168.10.21  TCP       [Retransmission] 502 → 54321
54   48.008400 192.168.10.100  192.168.10.21  TCP       [Retransmission] 502 → 54321
55   48.010500 192.168.10.100  192.168.10.21  TCP       [Retransmission] 502 → 54321
... (持续重传，无响应)

关键发现：PoE 断开后，设备以太网 PHY 失去链路，TCP 重传持续约 3 秒后客户端放弃，连接关闭。

3. PoE 恢复后的链路协商

No.  Time      Source          Destination     Protocol  Info
...  (PoE 恢复，链路重新协商)
120  52.500000                   Ethernet    LLDP / CDP 交换机发现
121  52.502000 192.168.10.21    broadcast   ARP         Who has 192.168.10.1?
122  52.503000 192.168.10.1     192.168.10.21 ARP         192.168.10.1 is at xx:xx:xx
123  52.505000 192.168.10.100  192.168.10.21  TCP       54322 → 502 [SYN]  (新连接)
124  52.505012 192.168.10.21   192.168.10.100 TCP       502 → 54322 [SYN, ACK]
125  52.505020 192.168.10.100  192.168.10.21  TCP       54322 → 502 [ACK]
126  52.505100 192.168.10.100  192.168.10.21  Modbus    Read Holding Registers
127  52.505180 192.168.10.21   192.168.10.100 Modbus    Response: temp=22.3, rh=55.9

时序分解：

阶段	耗时
PoE 断开 → 交换机端口 down	< 1s
交换机端口 down → 设备 PHY 失去链路	< 0.5s
设备 MCU 复位（如果有）	0–15s（取决于是否有超级电容）
PoE 恢复 → 链路协商完成	2–5s
TCP 重连	< 0.1s
Modbus 首次成功响应	< 0.1s
总中断时长	5–20s（取决于设备设计）

四、Python 测试脚本

1. 持续采集 + 中断检测

import asyncio
import time
import struct
from pymodbus.client import AsyncModbusTcpClient
from influxdb_client import InfluxDBClient, Point
import csv

class ContinuityTester:
    def __init__(self, host, port, slave_id, poll_interval=1):
        self.host = host
        self.port = port
        self.slave_id = slave_id
        self.poll_interval = poll_interval
        self.client = None
        self.connected = False
        self.results = []
        self.influx = InfluxDBClient(
            url="http://localhost:8086",
            token="your-token",
            org="archive"
        )
        self.write_api = self.influx.write_api()

    async def connect(self):
        try:
            self.client = AsyncModbusTcpClient(self.host, port=self.port)
            await self.client.connect()
            self.connected = self.client.connected
            return self.connected
        except Exception as e:
            print(f"[{time.time():.3f}] CONNECT FAILED: {e}")
            self.connected = False
            return False

    async def poll_once(self):
        """单次轮询，记录响应时间和质量"""
        ts = time.time()
        try:
            if not self.client or not self.client.connected:
                raise ConnectionError("Not connected")

            start = time.perf_counter()
            resp = await self.client.read_holding_registers(
                address=40001, count=2, slave=self.slave_id
            )
            elapsed_ms = (time.perf_counter() - start) * 1000

            if resp.isError() or not resp.registers:
                raise ValueError("Modbus error or empty response")

            temp = resp.registers[0] * 0.1
            humid = resp.registers[1] * 0.1

            result = {
                'timestamp': ts,
                'temp': temp,
                'humid': humid,
                'response_ms': elapsed_ms,
                'quality': 'good',
                'error': None
            }
            self.results.append(result)

            # 写入 InfluxDB
            point = Point("power_switch_test") \
                .tag("device", self.host) \
                .field("temperature", temp) \
                .field("humidity", humid) \
                .field("response_ms", elapsed_ms) \
                .field("quality", 0) \
                .time(int(ts * 1e9))
            self.write_api.write(bucket="env_monitor", record=point)

            print(f"[{ts:.3f}] OK  temp={temp:.1f} rh={humid:.1f}  rtt={elapsed_ms:.1f}ms")
            return result

        except Exception as e:
            result = {
                'timestamp': ts,
                'temp': None,
                'humid': None,
                'response_ms': None,
                'quality': 'bad',
                'error': str(e)
            }
            self.results.append(result)

            point = Point("power_switch_test") \
                .tag("device", self.host) \
                .field("quality", 2) \
                .field("response_ms", -1) \
                .time(int(ts * 1e9))
            self.write_api.write(bucket="env_monitor", record=point)

            print(f"[{ts:.3f}] FAIL  {str(e)[:60]}")
            return result

    async def run_test(self, duration_seconds=300):
        """持续测试指定时长"""
        print(f"Starting continuity test for {duration_seconds}s...")
        print(f"Target: {self.host}:{self.port}")
        print(f"Poll interval: {self.poll_interval}s")
        print("=" * 60)

        await self.connect()
        start_time = time.time()

        while (time.time() - start_time) < duration_seconds:
            await self.poll_once()

            # 如果连接断了，尝试重连
            if not self.client or not self.client.connected:
                print(f"[{time.time():.3f}] Attempting reconnect...")
                await asyncio.sleep(2)
                await self.connect()
            else:
                await asyncio.sleep(self.poll_interval)

        # 保存 CSV
        with open(f'test_results_{int(start_time)}.csv', 'w', newline='') as f:
            writer = csv.DictWriter(f, fieldnames=self.results[0].keys())
            writer.writeheader()
            writer.writerows(self.results)

        print(f"\nTest complete. {len(self.results)} samples recorded.")
        self.analyze_results()

    def analyze_results(self):
        """分析中断时长和数据连续性"""
        gaps = []
        last_good = None

        for i, r in enumerate(self.results):
            if r['quality'] == 'good':
                if last_good is not None:
                    gap = r['timestamp'] - self.results[last_good]['timestamp']
                    if gap > self.poll_interval * 2:
                        gaps.append({
                            'start': self.results[last_good]['timestamp'],
                            'end': r['timestamp'],
                            'duration': gap
                        })
                last_good = i

        print("\n" + "=" * 60)
        print("ANALYSIS RESULTS")
        print("=" * 60)
        print(f"Total samples:     {len(self.results)}")
        print(f"Good samples:       {sum(1 for r in self.results if r['quality'] == 'good')}")
        print(f"Bad samples:        {sum(1 for r in self.results if r['quality'] == 'bad')}")
        print(f"Interruptions:      {len(gaps)}")

        for i, g in enumerate(gaps):
            print(f"  Gap {i+1}: {g['duration']:.1f}s "
                  f"({time.strftime('%H:%M:%S', time.localtime(g['start']))} → "
                  f"{time.strftime('%H:%M:%S', time.localtime(g['end']))})")

        if gaps:
            max_gap = max(g['duration'] for g in gaps)
            print(f"\nMax interruption:   {max_gap:.1f}s")
            if max_gap > 10:
                print("⚠️  FAIL: Max interruption exceeds 10s threshold")
            else:
                print("✅ PASS: Max interruption within acceptable range")


# 运行测试
async def main():
    tester = ContinuityTester(
        host="192.168.10.21",
        port=502,
        slave_id=1,
        poll_interval=1
    )
    await tester.run_test(duration_seconds=600)  # 10 分钟测试


asyncio.run(main())

2. 远程控制电源切换（通过串口/网络控制可编程电源）

import serial
import time

class ProgrammablePowerSupply:
    """控制可编程直流电源通断"""

    def __init__(self, port, baudrate=9600):
        self.ser = serial.Serial(port, baudrate, timeout=1)
        self._initialize()

    def _initialize(self):
        """初始化电源（SCPI 命令）"""
        self.ser.write(b"*CLS\n")
        self.ser.write(b"OUTP:STAT ON\n")
        self.ser.write(b"VOLT 24\n")
        self.ser.write(b"CURR 2\n")
        time.sleep(0.5)

    def turn_on(self):
        self.ser.write(b"OUTP:STAT ON\n")
        print(f"[{time.time():.3f}] 24VDC ON")

    def turn_off(self):
        self.ser.write(b"OUTP:STAT OFF\n")
        print(f"[{time.time():.3f}] 24VDC OFF")

    def toggle(self, off_duration=5):
        """断开指定秒数后恢复"""
        self.turn_off()
        time.sleep(off_duration)
        self.turn_on()

五、InfluxDB 中的测试结果分析

1. 查询中断事件

-- 找出所有中断事件（连续两次 good 之间间隔 > 2s）
SELECT * FROM (
  SELECT
    LAST(quality) AS quality,
    time
  FROM env_monitor.raw_30d.power_switch_test
  WHERE device = '192.168.10.21'
  GROUP BY time(1s)
)

2. Grafana 面板配置

Panel 1: 温湿度曲线（绿色 = good，红色 = bad）
  SELECT mean(temperature) FROM power_switch_test WHERE quality = 0
  SELECT mean(humidity) FROM power_switch_test WHERE quality = 0

Panel 2: 响应时间（ms）
  SELECT mean(response_ms) FROM power_switch_test

Panel 3: 质量位（0=good, 2=bad）
  SELECT mean(quality) FROM power_switch_test

Panel 4: 中断事件标注
  SELECT count(quality) FROM power_switch_test WHERE quality = 2
  (用 annotation 标注)

六、不同设备的测试结果对比

设备型号	供电切换	中断时长	MCU 复位	恢复后 TCP 重连	评价
品牌 A（高端）	24VDC ↔ PoE	0.5s	否	自动	✅ 超级电容缓冲，无缝
品牌 B（中端）	24VDC → PoE	3–5s	否	自动	✅ 可接受
品牌 B（中端）	PoE → 24VDC	8–12s	是（部分型号）	自动	⚠️ 需优化
品牌 C（低端）	任意切换	15–30s	是	需要手动/重启	❌ 不合格
品牌 D（工业）	24VDC ↔ PoE	0s	否	不断	✅ 双电源独立 DC-DC

七、设计建议：如何选型和部署

选型阶段

检查项	要求
内部是否有超级电容/电池缓冲	必须有，缓冲时间 ≥ 10s
MCU 在供电切换时是否复位	不能复位
以太网 PHY 是否支持链路快速恢复	支持 IEEE 802.3az 快速链路
是否支持 TCP 连接保活	支持 Keepalive，建议 30s/3次
是否支持 MQTT 会话持久化	Clean Session = false

部署阶段

措施	说明
PoE 交换机配置静态供电	禁止 PoE 节能模式（端口休眠）
交换机端口配置为"永不关闭"	power inline never off
网关侧 TCP Keepalive	服务器端启用，探测间隔 30s
应用层心跳	每 10s 一次心跳，超时 3 次判定离线
本地缓存	SQLite WAL，断网/断电不丢数据

八、常见返工点

问题	后果	正确做法
标书只写"支持双供电"	低端设备也能满足字面要求	明确写入"切换中断 ≤ 3s，MCU 不复位"
不测切换时序	验收时发现问题，返工成本极高	到货即做切换测试，记录中断时长
PoE 交换机端口休眠	夜间低流量时端口关闭 PoE	配置静态 PoE，关闭节能
网关不重连	设备恢复后网关不重连 TCP	指数退避重连 + 持久会话
不标记质量位	切换期间的数据空洞被"平滑"	quality=bad 明确标记
用平均值掩盖中断	报表看起来正常，实际数据断了	保留 sample_count，中断期间 count=0

九、一句话总结

双供电冗余的价值不在于"有两路电"，而在于切换时数据不断、连接不丢、审计链完整。选型时别信彩页，部署前别省测试——用 Wireshark 抓一次包、用 Python 跑一次连续性测试，5 分钟就能知道这台设备到底是"真冗余"还是"假双供"。

当飞检人员问"如果 PoE 交换机故障，你的监测数据会不会断"时，你拿出的不应该是一份彩页，而应该是一份切换测试报告——上面有精确到毫秒的中断时长、TCP 重连时序、InfluxDB 中的质量位标记。这才是双供电冗余的真正证明，而不是"两条线都插着"的证明。

原创声明：本文系作者授权腾讯云开发者社区发表，未经许可，不得转载。

如有侵权，请联系 cloudcommunity@tencent.com 删除。

功能测试

#网口温湿度记录仪

24VDC 与 POE 双供电冗余设计：网口温湿度记录仪在供电切换时的数据连续性测试

24VDC 与 POE 双供电冗余设计：网口温湿度记录仪在供电切换时的数据连续性测试

24VDC 与 POE 双供电冗余设计：网口温湿度记录仪在供电切换时的数据连续性测试

一、双供电冗余的"理论"与"现实"

1. 硬件层面的冗余架构

二、测试方案设计

1. 测试目标

2. 测试拓扑

3. 测试工具

三、Wireshark 抓包分析：切换时到底发生了什么

1. 正常状态下的通信

2. PoE 断开瞬间

3. PoE 恢复后的链路协商

四、Python 测试脚本

1. 持续采集 + 中断检测

2. 远程控制电源切换（通过串口/网络控制可编程电源）

五、InfluxDB 中的测试结果分析

1. 查询中断事件

2. Grafana 面板配置

六、不同设备的测试结果对比

七、设计建议：如何选型和部署

选型阶段

部署阶段

八、常见返工点

九、一句话总结

社区

活动

圈层

关于

腾讯云开发者

热门产品

热门推荐

更多推荐