机房传感网络排障：以太网温湿度变送器 UDP 报文异常与断线重连优化

现象：某机房 24 台 UDP 温湿度变送器，运行 3 天后出现：
  - 监控平台随机丢失数据点（约 3-5% 的数据包）
  - 每天凌晨 2:00-2:15 期间丢包率飙升至 40%
  - 2 台设备连续 5 分钟无数据后自动恢复
  - 交换机端口无 error，ping 网关正常

第一层：应用层（数据是否正确产生）
  → 设备 LCD 显示正常，本地数据在更新
  → 排除传感器硬件故障

第二层：网络层（数据是否传输）
  → ping 通（ICMP 正常）
  → UDP 端口可达性测试：nc -u 测试正常
  → 但 tcpdump 抓包发现丢包

第三层：中间设备（是否有中间节点丢 UDP）
  → 逐跳 ping + UDP 打流测试
  → 定位到接入交换机上行端口存在 UDP 丢包

# 抓取指定传感器的 UDP 报文（假设端口 8888）
tcpdump -nn -i eth0 udp port 8888 and src host 10.20.30.41 -c 1000 -w udp_capture.pcap

# 用 Wireshark 分析
# 统计 → 会话 → UDP
# 发现：发送间隔应为 5s，实际抓包间隔不均匀
# 部分报文间隔 > 30s（说明中间有丢失）

正常模式：每 5s 一个包，间隔均匀
异常模式：
  09:00:00  seq=1000  ✓
  09:00:05  seq=1001  ✓
  09:00:10  seq=1002  ✗ (丢失)
  09:00:15  seq=1003  ✓
  09:00:20  seq=1004  ✗ (丢失)
  09:00:25  seq=1005  ✓

02:00  Zabbix 执行每日自动发现扫描（SNMP walk 全网段）
02:00  备份系统启动全量备份，占用管理网带宽
02:00  日志系统轮转，写入压力增大

# 在凌晨 2:00 前后测试 UDP 吞吐量
iperf3 -u -c 10.20.30.100 -b 10M -t 60
# 结果显示：02:00 时段 UDP 丢包率 35-45%，其他时段 < 5%

# Cisco 交换机
show interfaces gigabitEthernet 1/0/1 | include drops|errors
# 输出：
#   Input queue: 0/2000/0/0 (size/max/drops/flushes)
#   Total output drops: 12847    ← 输出丢包
#   Queueing strategy: fifo

# H3C 交换机
display interface GigabitEthernet 1/0/1
# 输出：
#  Input:  0 drops, 0 errors
#  Output: 12847 drops           ← 输出丢包

交换机端口速率：1Gbps
实际 UDP 流量：约 2Mbps（24 台 × 每包 200 字节 × 每 5s）
其他流量：管理流量、备份流量、SNMP 轮询

# 查看微突发
show interfaces gigabitEthernet 1/0/1 | include rate
# 输出：
#   Input rate 2.1 Mbps, Output rate 850 Mbps (峰值)
#   Peak output rate: 980 Mbps (持续 10-50ms)

发送端：seq=1001 → seq=1002 → seq=1003
接收端：seq=1001 → seq=1003 → seq=1002  (乱序)

class UDPReceiver:
    def __init__(self, window_size=10, timeout_ms=100):
        self.window = {}           # seq → (timestamp, data)
        self.window_size = window_size
        self.timeout_ms = timeout_ms
        self.expected_seq = 0

    def on_packet(self, seq, data, recv_time):
        # 丢弃太老的包
        if seq < self.expected_seq - self.window_size:
            return None

        self.window[seq] = (recv_time, data)

        # 尝试按序交付
        delivered = []
        while self.expected_seq in self.window:
            _, data = self.window.pop(self.expected_seq)
            delivered.append((self.expected_seq, data))
            self.expected_seq += 1

        # 清理过期包
        now = time.time() * 1000
        self.window = {
            k: v for k, v in self.window.items()
            if now - v[0] < self.timeout_ms
        }

        return delivered if delivered else None

接收端收到：seq=1001, seq=1001, seq=1002, seq=1001

class Deduplicator:
    def __init__(self, max_cache=1000):
        self.seen = {}          # seq → receive_count
        self.max_cache = max_cache

    def is_duplicate(self, seq):
        if seq in self.seen:
            self.seen[seq] += 1
            return True
        self.seen[seq] = 1
        # 清理旧条目
        if len(self.seen) > self.max_cache:
            oldest = min(self.seen.keys())
            del self.seen[oldest]
        return False

原始报文：1800 字节
IP 分片：碎片1 (1500B) + 碎片2 (300B)
碎片2 丢失 → 整包不可用

# 方案1：应用层控制报文大小
def build_packet(sensor_id, temp, humi, seq):
    """确保报文 < 1472 字节（1500 - IP头20B - UDP头8B）"""
    payload = struct.pack('>Bf f I',
                          sensor_id,
                          temp,      # 4 bytes
                          humi,      # 4 bytes
                          seq)       # 4 bytes
    # 总共 13 字节，远小于 MTU
    return payload

# 方案2：应用层分片 + 重组
class AppFragmenter:
    MAX_FRAGMENT = 1400  # 留余量

    def fragment(self, data, seq):
        """将大数据分片，每片带 seq + frag_id + total_frags"""
        frags = []
        for i in range(0, len(data), self.MAX_FRAGMENT):
            frag = struct.pack('>I H H', seq, i // self.MAX_FRAGMENT,
                               (len(data) + self.MAX_FRAGMENT - 1) // self.MAX_FRAGMENT)
            frag += data[i:i + self.MAX_FRAGMENT]
            frags.append(frag)
        return frags

import hashlib

def build_packet_with_crc(sensor_id, temp, humi, seq):
    payload = struct.pack('>B f f I', sensor_id, temp, humi, seq)
    crc = hashlib.md5(payload).digest()[:4]  # 4 字节 CRC
    return payload + crc

def verify_packet(data):
    if len(data) < 4:
        return False
    payload = data[:-4]
    expected_crc = data[-4:]
    actual_crc = hashlib.md5(payload).digest()[:4]
    return expected_crc == actual_crc

class HeartbeatManager:
    def __init__(self, interval=10, timeout=30):
        self.interval = interval      # 心跳间隔（秒）
        self.timeout = timeout        # 超时阈值（秒）
        self.last_seen = {}           # device_id → last_heartbeat_time
        self.callbacks = []

    def register_heartbeat(self, device_id, timestamp):
        self.last_seen[device_id] = timestamp

    def check_alive(self, device_id):
        if device_id not in self.last_seen:
            return False
        elapsed = time.time() - self.last_seen[device_id]
        return elapsed < self.timeout

    def monitor_loop(self):
        """定期检查所有设备在线状态"""
        while True:
            now = time.time()
            for device_id, last_time in list(self.last_seen.items()):
                if now - last_time > self.timeout:
                    # 设备离线
                    for cb in self.callbacks:
                        cb(device_id, "offline", now - last_time)
                    del self.last_seen[device_id]
            time.sleep(5)

    def on_offline(self, callback):
        self.callbacks.append(callback)

/* 设备端伪代码（C语言） */
typedef struct {
    int sockfd;
    struct sockaddr_in server_addr;
    uint32_t seq;
    uint8_t reconnect_count;
    uint32_t last_send_time;
} sensor_ctx_t;

void sensor_main_loop(sensor_ctx_t *ctx) {
    while (1) {
        /* 读取传感器数据 */
        float temp = read_temperature();
        float humi = read_humidity();

        /* 构造 UDP 报文 */
        uint8_t buf[32];
        int len = build_packet(buf, temp, humi, ctx->seq++);

        /* 发送 */
        ssize_t sent = sendto(ctx->sockfd, buf, len, 0,
                              (struct sockaddr*)&ctx->server_addr,
                              sizeof(ctx->server_addr));

        if (sent < 0) {
            /* 发送失败，尝试重连 */
            handle_send_error(ctx);
        } else {
            ctx->last_send_time = get_tick_ms();
            ctx->reconnect_count = 0;  /* 重置重连计数 */
        }

        /* 检查是否需要心跳 */
        if (get_tick_ms() - ctx->last_send_time > HEARTBEAT_INTERVAL * 1000) {
            send_heartbeat(ctx);
        }

        delay_ms(SEND_INTERVAL_MS);
    }
}

void handle_send_error(sensor_ctx_t *ctx) {
    ctx->reconnect_count++;

    /* 指数退避：1s, 2s, 4s, 8s, 16s, 最大 60s */
    uint32_t backoff = min(60, 1 << min(ctx->reconnect_count, 6));
    delay_ms(backoff * 1000);

    /* 关闭旧 socket */
    close(ctx->sockfd);

    /* 重新创建 socket */
    ctx->sockfd = socket(AF_INET, SOCK_DGRAM, 0);
    if (ctx->sockfd < 0) {
        /* socket 创建失败，继续重试 */
        return;
    }

    /* 可选：重新绑定端口（某些设备需要固定源端口） */
    struct sockaddr_in local_addr;
    memset(&local_addr, 0, sizeof(local_addr));
    local_addr.sin_family = AF_INET;
    local_addr.sin_addr.s_addr = INADDR_ANY;
    local_addr.sin_port = htons(LOCAL_PORT);

    bind(ctx->sockfd, (struct sockaddr*)&local_addr, sizeof(local_addr));
}

class DeviceRegistry:
    def __init__(self):
        self.devices = {}          # device_id → {ip, port, last_seen, status}
        self.pending_reconnect = {}  # device_id → retry_count

    def handle_packet(self, data, addr):
        """处理来自设备的任何报文"""
        device_id = parse_device_id(data)

        if device_id not in self.devices:
            # 新设备首次出现
            self.register_device(device_id, addr)
        elif self.devices[device_id]['ip'] != addr[0]:
            # IP 变化（DHCP 续租）
            self.update_device_address(device_id, addr)
        else:
            # 已知设备，更新最后通信时间
            self.devices[device_id]['last_seen'] = time.time()

    def register_device(self, device_id, addr):
        """新设备注册"""
        self.devices[device_id] = {
            'ip': addr[0],
            'port': addr[1],
            'first_seen': time.time(),
            'last_seen': time.time(),
            'status': 'online'
        }
        logger.info(f"New device registered: {device_id} at {addr[0]}:{addr[1]}")

        # 触发设备上线事件
        trigger_event('device_online', device_id, addr)

    def update_device_address(self, device_id, addr):
        """设备 IP 变化"""
        old_ip = self.devices[device_id]['ip']
        self.devices[device_id]['ip'] = addr[0]
        self.devices[device_id]['port'] = addr[1]
        self.devices[device_id]['last_seen'] = time.time()
        logger.warning(f"Device {device_id} IP changed: {old_ip} → {addr[0]}")

        # 触发 IP 变化事件（可能需要重新配置防火墙规则等）
        trigger_event('device_ip_changed', device_id, old_ip, addr[0])

Cisco 交换机示例：

! 创建 ACL 匹配传感 UDP 流量
ip access-list extended SENSOR_UDP
 permit udp any any eq 8888

! 创建 class-map
class-map match-all SENSOR_CLASS
 match access-group name SENSOR_UDP

! 创建 policy-map
policy-map QOS_POLICY
 class SENSOR_CLASS
  priority level 1          ! 最高优先级
  police 5000000 100000     ! 限速 5Mbps，突发 100KB

! 应用到端口
interface GigabitEthernet1/0/1
 service-policy output QOS_POLICY

# Linux 防火墙 UDP 超时（默认 30s）
sysctl -w net.netfilter.nf_conntrack_udp_timeout=120
sysctl -w net.netfilter.nf_conntrack_udp_timeout_stream=180

# 持久化
echo "net.netfilter.nf_conntrack_udp_timeout=120" >> /etc/sysctl.conf

方案1：错峰上报
  - 传感器随机化上报间隔（5±1s），避免同步发送
  - 或按设备 ID 哈希分配上报相位

方案2：凌晨降低上报频率
  - 设备端根据时间自动调整：白天 5s，凌晨 2:00-4:00 改为 30s

方案3：管理网带宽预留
  - 为传感流量预留 10Mbps 保障带宽

# 服务器端
class UDPServerWithACK:
    def __init__(self, host, port):
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.sock.bind((host, port))
        self.pending = {}  # (device_id, seq) → send_time
        self.ack_timeout = 3  # 秒

    def handle_data(self, data, addr):
        device_id, seq, temp, humi = parse_packet(data)

        # 发送 ACK
        ack = struct.pack('>B I', 0x01, seq)  # 0x01 = ACK
        self.sock.sendto(ack, addr)

        # 处理数据
        process_data(device_id, temp, humi)

    def run(self):
        while True:
            data, addr = self.sock.recvfrom(1500)
            self.handle_data(data, addr)

# 设备端
class SensorWithACK:
    def __init__(self):
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.sock.settimeout(2)  # 2 秒超时

    def send_with_ack(self, data, server_addr, max_retries=3):
        for attempt in range(max_retries):
            self.sock.sendto(data, server_addr)
            try:
                ack, _ = self.sock.recvfrom(1024)
                if verify_ack(ack, data):
                    return True  # 发送成功
            except socket.timeout:
                continue  # 超时重传
        return False  # 所有重试失败

def batch_report(sensor_id, readings, seq_base):
    """
    readings: [(temp, humi, timestamp), ...] 最多 10 个点
    压缩后单包发送
    """
    buf = struct.pack('>B I B', sensor_id, seq_base, len(readings))
    for temp, humi, ts in readings:
        buf += struct.pack('>f f I', temp, humi, int(ts))

    # 如果超过 MTU，拆分为多个包
    if len(buf) > 1400:
        return split_batch(sensor_id, readings, seq_base)
    return [buf]

现象：某机房 24 台 UDP 温湿度变送器，运行 3 天后出现：
  - 监控平台随机丢失数据点（约 3-5% 的数据包）
  - 每天凌晨 2:00-2:15 期间丢包率飙升至 40%
  - 2 台设备连续 5 分钟无数据后自动恢复
  - 交换机端口无 error，ping 网关正常

第一层：应用层（数据是否正确产生）
  → 设备 LCD 显示正常，本地数据在更新
  → 排除传感器硬件故障

第二层：网络层（数据是否传输）
  → ping 通（ICMP 正常）
  → UDP 端口可达性测试：nc -u 测试正常
  → 但 tcpdump 抓包发现丢包

第三层：中间设备（是否有中间节点丢 UDP）
  → 逐跳 ping + UDP 打流测试
  → 定位到接入交换机上行端口存在 UDP 丢包

# 抓取指定传感器的 UDP 报文（假设端口 8888）
tcpdump -nn -i eth0 udp port 8888 and src host 10.20.30.41 -c 1000 -w udp_capture.pcap

# 用 Wireshark 分析
# 统计 → 会话 → UDP
# 发现：发送间隔应为 5s，实际抓包间隔不均匀
# 部分报文间隔 > 30s（说明中间有丢失）

正常模式：每 5s 一个包，间隔均匀
异常模式：
  09:00:00  seq=1000  ✓
  09:00:05  seq=1001  ✓
  09:00:10  seq=1002  ✗ (丢失)
  09:00:15  seq=1003  ✓
  09:00:20  seq=1004  ✗ (丢失)
  09:00:25  seq=1005  ✓

02:00  Zabbix 执行每日自动发现扫描（SNMP walk 全网段）
02:00  备份系统启动全量备份，占用管理网带宽
02:00  日志系统轮转，写入压力增大

# 在凌晨 2:00 前后测试 UDP 吞吐量
iperf3 -u -c 10.20.30.100 -b 10M -t 60
# 结果显示：02:00 时段 UDP 丢包率 35-45%，其他时段 < 5%

# Cisco 交换机
show interfaces gigabitEthernet 1/0/1 | include drops|errors
# 输出：
#   Input queue: 0/2000/0/0 (size/max/drops/flushes)
#   Total output drops: 12847    ← 输出丢包
#   Queueing strategy: fifo

# H3C 交换机
display interface GigabitEthernet 1/0/1
# 输出：
#  Input:  0 drops, 0 errors
#  Output: 12847 drops           ← 输出丢包

交换机端口速率：1Gbps
实际 UDP 流量：约 2Mbps（24 台 × 每包 200 字节 × 每 5s）
其他流量：管理流量、备份流量、SNMP 轮询

# 查看微突发
show interfaces gigabitEthernet 1/0/1 | include rate
# 输出：
#   Input rate 2.1 Mbps, Output rate 850 Mbps (峰值)
#   Peak output rate: 980 Mbps (持续 10-50ms)

发送端：seq=1001 → seq=1002 → seq=1003
接收端：seq=1001 → seq=1003 → seq=1002  (乱序)

class UDPReceiver:
    def __init__(self, window_size=10, timeout_ms=100):
        self.window = {}           # seq → (timestamp, data)
        self.window_size = window_size
        self.timeout_ms = timeout_ms
        self.expected_seq = 0

    def on_packet(self, seq, data, recv_time):
        # 丢弃太老的包
        if seq < self.expected_seq - self.window_size:
            return None

        self.window[seq] = (recv_time, data)

        # 尝试按序交付
        delivered = []
        while self.expected_seq in self.window:
            _, data = self.window.pop(self.expected_seq)
            delivered.append((self.expected_seq, data))
            self.expected_seq += 1

        # 清理过期包
        now = time.time() * 1000
        self.window = {
            k: v for k, v in self.window.items()
            if now - v[0] < self.timeout_ms
        }

        return delivered if delivered else None

接收端收到：seq=1001, seq=1001, seq=1002, seq=1001

class Deduplicator:
    def __init__(self, max_cache=1000):
        self.seen = {}          # seq → receive_count
        self.max_cache = max_cache

    def is_duplicate(self, seq):
        if seq in self.seen:
            self.seen[seq] += 1
            return True
        self.seen[seq] = 1
        # 清理旧条目
        if len(self.seen) > self.max_cache:
            oldest = min(self.seen.keys())
            del self.seen[oldest]
        return False

原始报文：1800 字节
IP 分片：碎片1 (1500B) + 碎片2 (300B)
碎片2 丢失 → 整包不可用

# 方案1：应用层控制报文大小
def build_packet(sensor_id, temp, humi, seq):
    """确保报文 < 1472 字节（1500 - IP头20B - UDP头8B）"""
    payload = struct.pack('>Bf f I',
                          sensor_id,
                          temp,      # 4 bytes
                          humi,      # 4 bytes
                          seq)       # 4 bytes
    # 总共 13 字节，远小于 MTU
    return payload

# 方案2：应用层分片 + 重组
class AppFragmenter:
    MAX_FRAGMENT = 1400  # 留余量

    def fragment(self, data, seq):
        """将大数据分片，每片带 seq + frag_id + total_frags"""
        frags = []
        for i in range(0, len(data), self.MAX_FRAGMENT):
            frag = struct.pack('>I H H', seq, i // self.MAX_FRAGMENT,
                               (len(data) + self.MAX_FRAGMENT - 1) // self.MAX_FRAGMENT)
            frag += data[i:i + self.MAX_FRAGMENT]
            frags.append(frag)
        return frags

import hashlib

def build_packet_with_crc(sensor_id, temp, humi, seq):
    payload = struct.pack('>B f f I', sensor_id, temp, humi, seq)
    crc = hashlib.md5(payload).digest()[:4]  # 4 字节 CRC
    return payload + crc

def verify_packet(data):
    if len(data) < 4:
        return False
    payload = data[:-4]
    expected_crc = data[-4:]
    actual_crc = hashlib.md5(payload).digest()[:4]
    return expected_crc == actual_crc

class HeartbeatManager:
    def __init__(self, interval=10, timeout=30):
        self.interval = interval      # 心跳间隔（秒）
        self.timeout = timeout        # 超时阈值（秒）
        self.last_seen = {}           # device_id → last_heartbeat_time
        self.callbacks = []

    def register_heartbeat(self, device_id, timestamp):
        self.last_seen[device_id] = timestamp

    def check_alive(self, device_id):
        if device_id not in self.last_seen:
            return False
        elapsed = time.time() - self.last_seen[device_id]
        return elapsed < self.timeout

    def monitor_loop(self):
        """定期检查所有设备在线状态"""
        while True:
            now = time.time()
            for device_id, last_time in list(self.last_seen.items()):
                if now - last_time > self.timeout:
                    # 设备离线
                    for cb in self.callbacks:
                        cb(device_id, "offline", now - last_time)
                    del self.last_seen[device_id]
            time.sleep(5)

    def on_offline(self, callback):
        self.callbacks.append(callback)

/* 设备端伪代码（C语言） */
typedef struct {
    int sockfd;
    struct sockaddr_in server_addr;
    uint32_t seq;
    uint8_t reconnect_count;
    uint32_t last_send_time;
} sensor_ctx_t;

void sensor_main_loop(sensor_ctx_t *ctx) {
    while (1) {
        /* 读取传感器数据 */
        float temp = read_temperature();
        float humi = read_humidity();

        /* 构造 UDP 报文 */
        uint8_t buf[32];
        int len = build_packet(buf, temp, humi, ctx->seq++);

        /* 发送 */
        ssize_t sent = sendto(ctx->sockfd, buf, len, 0,
                              (struct sockaddr*)&ctx->server_addr,
                              sizeof(ctx->server_addr));

        if (sent < 0) {
            /* 发送失败，尝试重连 */
            handle_send_error(ctx);
        } else {
            ctx->last_send_time = get_tick_ms();
            ctx->reconnect_count = 0;  /* 重置重连计数 */
        }

        /* 检查是否需要心跳 */
        if (get_tick_ms() - ctx->last_send_time > HEARTBEAT_INTERVAL * 1000) {
            send_heartbeat(ctx);
        }

        delay_ms(SEND_INTERVAL_MS);
    }
}

void handle_send_error(sensor_ctx_t *ctx) {
    ctx->reconnect_count++;

    /* 指数退避：1s, 2s, 4s, 8s, 16s, 最大 60s */
    uint32_t backoff = min(60, 1 << min(ctx->reconnect_count, 6));
    delay_ms(backoff * 1000);

    /* 关闭旧 socket */
    close(ctx->sockfd);

    /* 重新创建 socket */
    ctx->sockfd = socket(AF_INET, SOCK_DGRAM, 0);
    if (ctx->sockfd < 0) {
        /* socket 创建失败，继续重试 */
        return;
    }

    /* 可选：重新绑定端口（某些设备需要固定源端口） */
    struct sockaddr_in local_addr;
    memset(&local_addr, 0, sizeof(local_addr));
    local_addr.sin_family = AF_INET;
    local_addr.sin_addr.s_addr = INADDR_ANY;
    local_addr.sin_port = htons(LOCAL_PORT);

    bind(ctx->sockfd, (struct sockaddr*)&local_addr, sizeof(local_addr));
}

class DeviceRegistry:
    def __init__(self):
        self.devices = {}          # device_id → {ip, port, last_seen, status}
        self.pending_reconnect = {}  # device_id → retry_count

    def handle_packet(self, data, addr):
        """处理来自设备的任何报文"""
        device_id = parse_device_id(data)

        if device_id not in self.devices:
            # 新设备首次出现
            self.register_device(device_id, addr)
        elif self.devices[device_id]['ip'] != addr[0]:
            # IP 变化（DHCP 续租）
            self.update_device_address(device_id, addr)
        else:
            # 已知设备，更新最后通信时间
            self.devices[device_id]['last_seen'] = time.time()

    def register_device(self, device_id, addr):
        """新设备注册"""
        self.devices[device_id] = {
            'ip': addr[0],
            'port': addr[1],
            'first_seen': time.time(),
            'last_seen': time.time(),
            'status': 'online'
        }
        logger.info(f"New device registered: {device_id} at {addr[0]}:{addr[1]}")

        # 触发设备上线事件
        trigger_event('device_online', device_id, addr)

    def update_device_address(self, device_id, addr):
        """设备 IP 变化"""
        old_ip = self.devices[device_id]['ip']
        self.devices[device_id]['ip'] = addr[0]
        self.devices[device_id]['port'] = addr[1]
        self.devices[device_id]['last_seen'] = time.time()
        logger.warning(f"Device {device_id} IP changed: {old_ip} → {addr[0]}")

        # 触发 IP 变化事件（可能需要重新配置防火墙规则等）
        trigger_event('device_ip_changed', device_id, old_ip, addr[0])

Cisco 交换机示例：

! 创建 ACL 匹配传感 UDP 流量
ip access-list extended SENSOR_UDP
 permit udp any any eq 8888

! 创建 class-map
class-map match-all SENSOR_CLASS
 match access-group name SENSOR_UDP

! 创建 policy-map
policy-map QOS_POLICY
 class SENSOR_CLASS
  priority level 1          ! 最高优先级
  police 5000000 100000     ! 限速 5Mbps，突发 100KB

! 应用到端口
interface GigabitEthernet1/0/1
 service-policy output QOS_POLICY

# Linux 防火墙 UDP 超时（默认 30s）
sysctl -w net.netfilter.nf_conntrack_udp_timeout=120
sysctl -w net.netfilter.nf_conntrack_udp_timeout_stream=180

# 持久化
echo "net.netfilter.nf_conntrack_udp_timeout=120" >> /etc/sysctl.conf

方案1：错峰上报
  - 传感器随机化上报间隔（5±1s），避免同步发送
  - 或按设备 ID 哈希分配上报相位

方案2：凌晨降低上报频率
  - 设备端根据时间自动调整：白天 5s，凌晨 2:00-4:00 改为 30s

方案3：管理网带宽预留
  - 为传感流量预留 10Mbps 保障带宽

# 服务器端
class UDPServerWithACK:
    def __init__(self, host, port):
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.sock.bind((host, port))
        self.pending = {}  # (device_id, seq) → send_time
        self.ack_timeout = 3  # 秒

    def handle_data(self, data, addr):
        device_id, seq, temp, humi = parse_packet(data)

        # 发送 ACK
        ack = struct.pack('>B I', 0x01, seq)  # 0x01 = ACK
        self.sock.sendto(ack, addr)

        # 处理数据
        process_data(device_id, temp, humi)

    def run(self):
        while True:
            data, addr = self.sock.recvfrom(1500)
            self.handle_data(data, addr)

# 设备端
class SensorWithACK:
    def __init__(self):
        self.sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        self.sock.settimeout(2)  # 2 秒超时

    def send_with_ack(self, data, server_addr, max_retries=3):
        for attempt in range(max_retries):
            self.sock.sendto(data, server_addr)
            try:
                ack, _ = self.sock.recvfrom(1024)
                if verify_ack(ack, data):
                    return True  # 发送成功
            except socket.timeout:
                continue  # 超时重传
        return False  # 所有重试失败

def batch_report(sensor_id, readings, seq_base):
    """
    readings: [(temp, humi, timestamp), ...] 最多 10 个点
    压缩后单包发送
    """
    buf = struct.pack('>B I B', sensor_id, seq_base, len(readings))
    for temp, humi, ts in readings:
        buf += struct.pack('>f f I', temp, humi, int(ts))

    # 如果超过 MTU，拆分为多个包
    if len(buf) > 1400:
        return split_batch(sensor_id, readings, seq_base)
    return [buf]