TCP与UDP协议对比：机房RJ45温湿度传感器通信方案解析

网关主动发起请求 → 传感器响应数据

TCP 方式：
  网关与每台传感器建立 TCP 连接
  发送读取命令（Modbus TCP 功能码 0x03）
  等待响应，解析数据
  连接可复用或每次新建

UDP 方式：
  网关向传感器 UDP 端口发送请求包
  传感器返回数据报
  无连接，每个请求独立

传感器定时主动发送数据 → 网关监听端口接收

TCP 方式：
  传感器作为 Client 连接网关
  建立长连接后持续推送
  断线自动重连

UDP 方式：
  传感器向网关 IP:Port 发送数据报
  无需建立连接，定时发送
  丢包不重传（除非应用层实现）

class TcpPoller:
    def __init__(self, sensors: list, pool_size=16):
        self.sensors = sensors
        self.semaphore = asyncio.Semaphore(pool_size)
        self.timeout = 5

    async def poll_one(self, sensor: dict):
        async with self.semaphore:
            try:
                reader, writer = await asyncio.wait_for(
                    asyncio.open_connection(sensor['ip'], sensor.get('port', 502)),
                    timeout=3
                )
                # 发送 Modbus TCP 读取请求
                req = build_modbus_read_request(sensor.get('slave', 1), 0, 2)
                writer.write(req)
                await writer.drain()

                resp = await asyncio.wait_for(
                    reader.read(256), timeout=self.timeout
                )
                writer.close()
                await writer.wait_closed()

                if len(resp) < 9:
                    return {**sensor, 'status': 'SHORT_RESP'}

                temp = struct.unpack('>h', resp[9:11])[0] / 10
                hum = struct.unpack('>h', resp[11:13])[0] / 10
                return {**sensor, 'temp': temp, 'hum': hum, 'status': 'OK'}
            except asyncio.TimeoutError:
                return {**sensor, 'status': 'TIMEOUT'}
            except Exception as e:
                return {**sensor, 'status': 'ERR', 'msg': str(e)}

    async def poll_all(self):
        tasks = [self.poll_one(s) for s in self.sensors]
        return await asyncio.gather(*tasks)

class TcpConnectionPool:
    def __init__(self, max_idle=300):
        self._pools = {}          # ip -> (reader, writer, last_used)
        self.max_idle = max_idle

    async def get_connection(self, ip: str, port: int = 502):
        now = time.time()
        if ip in self._pools:
            reader, writer, last = self._pools[ip]
            if now - last < self.max_idle:
                self._pools[ip] = (reader, writer, now)
                return reader, writer
            else:
                writer.close()

        reader, writer = await asyncio.open_connection(ip, port)
        self._pools[ip] = (reader, writer, now)
        return reader, writer

    async def cleanup(self):
        now = time.time()
        for ip, (reader, writer, last) in list(self._pools.items()):
            if now - last > self.max_idle:
                writer.close()
                del self._pools[ip]

class UdpReceiver:
    def __init__(self, listen_port: int):
        self.port = listen_port
        self.buffer = {}          # device_id -> latest reading
        self.transport = None

    async def start(self):
        loop = asyncio.get_event_loop()
        self.transport, _ = await loop.create_datagram_endpoint(
            lambda: UdpProtocol(self),
            local_addr=('0.0.0.0', self.port)
        )

    def on_packet(self, data: bytes, addr: tuple):
        # 解析传感器 UDP 数据报
        # 格式：设备ID(4B) + 温度(float) + 湿度(float) + 序列号(2B)
        if len(data) < 12:
            return
        device_id = struct.unpack('>I', data[:4])[0]
        temp = struct.unpack('>f', data[4:8])[0]
        hum = struct.unpack('>f', data[8:12])[0]
        seq = struct.unpack('>H', data[12:14])[0] if len(data) >= 14 else 0

        self.buffer[device_id] = {
            'temp': temp,
            'hum': hum,
            'seq': seq,
            'addr': addr,
            'ts': time.time()
        }

    def get_latest(self, device_id: int = None):
        if device_id:
            return self.buffer.get(device_id)
        return self.buffer


class UdpProtocol(asyncio.DatagramProtocol):
    def __init__(self, receiver: UdpReceiver):
        self.receiver = receiver

    def datagram_received(self, data, addr):
        self.receiver.on_packet(data, addr)

class UdpReliableSender:
    """UDP 发送端实现简单重传机制"""

    def __init__(self, remote_addr: tuple, retry=3, timeout=2):
        self.remote = remote_addr
        self.retry = retry
        self.timeout = timeout
        self.seq = 0
        self._transport = None

    async def send_with_ack(self, payload: bytes):
        """发送数据并等待 ACK"""
        self.seq += 1
        pkt = struct.pack('>H', self.seq) + payload

        for attempt in range(self.retry):
            self._transport.sendto(pkt, self.remote)
            try:
                # 等待 ACK（简化：通过接收回调匹配 seq）
                ack = await asyncio.wait_for(
                    self._wait_ack(self.seq), timeout=self.timeout
                )
                return True
            except asyncio.TimeoutError:
                continue
        return False

    async def _wait_ack(self, seq: int):
        # 实际实现需维护等待队列
        pass

┌─────────────────────────────────────────────────────────────────┐
│  控制面（TCP）           │ 数据面（UDP）                       │
├─────────────────────────┼─────────────────────────────────────┤
│  • 设备配置/参数修改    │  • 温湿度周期上报                   │
│  • 启停指令下发         │  • 状态心跳                         │
│  • 告警事件             │  • 批量广播发现                     │
│  • 固件升级             │  • 非关键区高频采集                 │
│  • 阈值配置             │                                     │
└─────────────────────────┴─────────────────────────────────────┘

推荐组合：
  关键区（配电室/特藏库） → TCP 全量采集 + UDP 心跳辅助
  普通区（走廊/缓冲区）   → UDP 主动上报 + TCP 按需查询
  控制指令               → 永远走 TCP
  告警上报               → TCP + SMS 双通道

┌──────────────────┐
                    │   边缘网关        │
                    │                  │
                    │  TCP Server     │←── 控制指令下发
                    │  (Port 502)     │
                    │                  │
                    │  UDP Server     │←── 传感器主动上报
                    │  (Port 9000)    │
                    │                  │
                    │  决策引擎        │
                    │  (数据融合)      │
                    └────────┬─────────┘
                             │
              ┌──────────────┼──────────────┐
              ▼              ▼              ▼
        ┌──────────┐  ┌──────────┐  ┌──────────┐
        │ 关键区    │  │ 普通区    │  │ 缓冲区    │
        │ TCP 轮询  │  │ UDP 上报  │  │ UDP 上报  │
        │ 16台      │  │ 32台      │  │ 8台       │
        └──────────┘  └──────────┘  └──────────┘

class HybridCollector:
    def __init__(self, tcp_sensors: list, udp_port: int):
        self.tcp_poller = TcpPoller(tcp_sensors)
        self.udp_receiver = UdpReceiver(udp_port)
        self.results = {}

    async def start_udp(self):
        await self.udp_receiver.start()

    async def collect_cycle(self):
        # TCP 轮询关键区
        tcp_results = await self.tcp_poller.poll_all()
        for r in tcp_results:
            self.results[r['id']] = {**r, 'protocol': 'tcp'}

        # UDP 收集普通区
        udp_data = self.udp_receiver.get_latest()
        for dev_id, data in udp_data.items():
            if time.time() - data['ts'] < 60:   # 60s 内有效
                self.results[f"UDP-{dev_id}"] = {
                    **data, 'protocol': 'udp', 'status': 'OK'
                }
            else:
                self.results[f"UDP-{dev_id}"]['status'] = 'STALE'

        return self.results