高并发采集：同时轮询数百台 POE 供电的以太网温湿度变送器（Modbus TCP/IP）

┌─────────────────────────────────────────────────────────────┐
│                    采集服务（单进程 asyncio）                  │
│                                                             │
│  ┌─────────────────────────────────────────────────────┐    │
│  │              设备管理器（DeviceManager）               │    │
│  │  · 维护设备列表（IP、端口、从站地址、采集周期）        │    │
│  │  · 按区域/库房分组                                    │    │
│  │  · 健康状态跟踪（在线/离线/响应时间）                  │    │
│  └───────────────────┬─────────────────────────────────┘    │
│                      │                                      │
│  ┌───────────────────▼─────────────────────────────────┐    │
│  │              连接池（ConnectionPool）                   │    │
│  │  · 每台设备一个持久 TCP 连接（或按需创建）              │    │
│  │  · 连接健康检测（心跳/超时）                           │    │
│  │  · 自动重连（指数退避）                               │    │
│  │  · 并发限制（Semaphore 控制同时活跃请求数）             │    │
│  └───────────────────┬─────────────────────────────────┘    │
│                      │                                      │
│  ┌───────────────────▼─────────────────────────────────┐    │
│  │              采集调度器（Scheduler）                    │    │
│  │  · 按设备采集周期调度（非阻塞）                        │    │
│  │  · 错峰采集（避免同时发起所有请求）                     │    │
│  │  · 优先级（关键库房优先）                              │    │
│  └───────────────────┬─────────────────────────────────┘    │
│                      │                                      │
│  ┌───────────────────▼─────────────────────────────────┐    │
│  │              数据管道（DataPipeline）                   │    │
│  │  · 质量位标记（good/bad/timeout）                     │    │
│  │  · 变化过滤（deadband）                               │    │
│  │  · 批量写入 InfluxDB / 转发 Kafka                     │    │
│  └─────────────────────────────────────────────────────┘    │
└─────────────────────────────────────────────────────────────┘

import asyncio
import time
from dataclasses import dataclass, field
from typing import Dict, Optional
from pymodbus.client import AsyncModbusTcpClient
from pymodbus.exceptions import ModbusIOException, ConnectionException

@dataclass
class DeviceConfig:
    host: str
    port: int = 502
    slave_id: int = 1
    poll_interval: float = 5.0
    timeout: float = 2.0
    retry_limit: int = 3

@dataclass
class DeviceState:
    client: Optional[AsyncModbusTcpClient] = None
    connected: bool = False
    last_poll: float = 0.0
    last_success: float = 0.0
    consecutive_failures: int = 0
    response_times: list = field(default_factory=list)
    quality: str = "unknown"

class ConnectionPool:
    def __init__(self, max_concurrent: int = 100):
        self.devices: Dict[str, DeviceState] = {}
        self.configs: Dict[str, DeviceConfig] = {}
        self.semaphore = asyncio.Semaphore(max_concurrent)
        self._lock = asyncio.Lock()

    def add_device(self, name: str, config: DeviceConfig):
        self.configs[name] = config
        self.devices[name] = DeviceState()

    async def get_client(self, name: str) -> Optional[AsyncModbusTcpClient]:
        """获取或创建设备的 Modbus TCP 客户端"""
        state = self.devices[name]
        config = self.configs[name]

        if state.client is not None and state.connected:
            return state.client

        # 需要新建连接
        async with self._lock:
            # 双重检查
            if state.client is not None and state.connected:
                return state.client

            # 关闭旧连接
            if state.client is not None:
                try:
                    state.client.close()
                except Exception:
                    pass

            state.client = AsyncModbusTcpClient(
                config.host,
                port=config.port,
                timeout=config.timeout,
                retries=1,
                retry_on_empty=True,
            )

            try:
                await state.client.connect()
                state.connected = state.client.connected
                if state.connected:
                    state.consecutive_failures = 0
                    return state.client
                else:
                    state.connected = False
                    return None
            except Exception as e:
                state.connected = False
                return None

    async def release(self, name: str):
        """释放信号量（在请求完成后调用）"""
        pass  # 信号量在 poll_device 中管理

    async def close_all(self):
        """关闭所有连接"""
        for name, state in self.devices.items():
            if state.client is not None:
                try:
                    state.client.close()
                except Exception:
                    pass
                state.client = None
                state.connected = False

class Scheduler:
    def __init__(self, pool: ConnectionPool, influx_writer=None):
        self.pool = pool
        self.influx = influx_writer
        self.running = False
        self._tasks: Dict[str, asyncio.Task] = {}

    async def start(self):
        """启动所有设备的采集任务"""
        self.running = True
        for name in self.pool.configs.keys():
            self._tasks[name] = asyncio.create_task(
                self._device_loop(name)
            )
        # 等待所有任务
        await asyncio.gather(*self._tasks.values(), return_exceptions=True)

    async def stop(self):
        """停止所有采集任务"""
        self.running = False
        for task in self._tasks.values():
            task.cancel()
        await asyncio.gather(*self._tasks.values(), return_exceptions=True)
        await self.pool.close_all()

    async def _device_loop(self, name: str):
        """单个设备的采集循环"""
        config = self.pool.configs[name]
        state = self.pool.devices[name]

        # 错峰启动：随机抖动 0-5 秒
        await asyncio.sleep(hash(name) % 5)

        while self.running:
            try:
                await self._poll_device(name)
            except asyncio.CancelledError:
                break
            except Exception as e:
                # 记录异常但不退出循环
                pass

            # 等待下一个采集周期
            await asyncio.sleep(config.poll_interval)

    async def _poll_device(self, name: str):
        """执行单次采集"""
        config = self.pool.configs[name]
        state = self.pool.devices[name]

        async with self.pool.semaphore:  # 限制并发数
            client = await self.pool.get_client(name)
            if client is None or not client.connected:
                state.quality = "bad"
                state.consecutive_failures += 1
                return

            t0 = time.monotonic()
            try:
                # 读取保持寄存器 40001-40002（温度、湿度）
                resp = await asyncio.wait_for(
                    client.read_holding_registers(
                        address=0, count=2, slave=config.slave_id
                    ),
                    timeout=config.timeout
                )

                elapsed = (time.monotonic() - t0) * 1000  # ms

                if resp is None or resp.isError():
                    raise ModbusIOException(f"Bad response: {resp}")

                # 解析数据
                temp = resp.registers[0] * 0.1
                humid = resp.registers[1] * 0.1

                # 更新状态
                state.last_poll = time.time()
                state.last_success = time.time()
                state.consecutive_failures = 0
                state.response_times.append(elapsed)
                if len(state.response_times) > 100:
                    state.response_times.pop(0)
                state.quality = "good"

                # 写入 InfluxDB
                if self.influx:
                    await self.influx.write_point(
                        measurement="temperature_humidity",
                        tags={"device": name, "host": config.host},
                        fields={
                            "temperature": temp,
                            "humidity": humid,
                            "response_ms": elapsed,
                            "quality": 0,
                        }
                    )

            except asyncio.TimeoutError:
                state.consecutive_failures += 1
                state.quality = "bad"
            except (ModbusIOException, ConnectionException, OSError) as e:
                state.consecutive_failures += 1
                state.quality = "bad"
                # 连接可能已断开，标记重连
                state.connected = False
                try:
                    client.close()
                except Exception:
                    pass
                state.client = None

from influxdb_client.client.influxdb_client_async import InfluxDBClientAsync

class AsyncInfluxWriter:
    def __init__(self, url: str, token: str, org: str, bucket: str):
        self.client = InfluxDBClientAsync(url=url, token=token, org=org)
        self.bucket = bucket
        self._queue = asyncio.Queue(maxsize=10000)
        self._task = None

    async def start(self):
        self._task = asyncio.create_task(self._flush_loop())

    async def stop(self):
        if self._task:
            self._task.cancel()
            await asyncio.gather(self._task, return_exceptions=True)
        await self.client.close()

    async def write_point(self, measurement: str, tags: dict, fields: dict):
        """非阻塞写入队列"""
        point = {
            "measurement": measurement,
            "tags": tags,
            "fields": fields,
            "time": int(time.time() * 1e9),
        }
        try:
            self._queue.put_nowait(point)
        except asyncio.QueueFull:
            # 队列满，丢弃最旧的数据
            try:
                self._queue.get_nowait()
            except asyncio.QueueEmpty:
                pass
            self._queue.put_nowait(point)

    async def _flush_loop(self):
        """批量写入 InfluxDB"""
        batch = []
        last_flush = time.monotonic()

        while True:
            try:
                # 等待数据，超时则刷新
                point = await asyncio.wait_for(self._queue.get(), timeout=1.0)
                batch.append(point)

                # 批量条件：达到 500 条或 5 秒
                if len(batch) >= 500 or (time.monotonic() - last_flush) >= 5.0:
                    await self._flush(batch)
                    batch.clear()
                    last_flush = time.monotonic()

            except asyncio.TimeoutError:
                if batch:
                    await self._flush(batch)
                    batch.clear()
                    last_flush = time.monotonic()
            except asyncio.CancelledError:
                if batch:
                    await self._flush(batch)
                break

    async def _flush(self, points: list):
        """批量写入 InfluxDB"""
        from influxdb_client import Point
        influx_points = []
        for p in points:
            pt = Point(p["measurement"]).time(p["time"])
            for k, v in p["tags"].items():
                pt = pt.tag(k, v)
            for k, v in p["fields"].items():
                pt = pt.field(k, v)
            influx_points.append(pt)

        try:
            await self.client.write_api().write(
                bucket=self.bucket, record=influx_points
            )
        except Exception as e:
            # 写入失败，记录日志，不重试（避免阻塞）
            pass

async def main():
    # 从配置文件加载设备列表
    devices = load_devices_from_config("devices.yaml")

    # 创建连接池
    pool = ConnectionPool(max_concurrent=100)

    # 添加设备
    for name, cfg in devices.items():
        pool.add_device(name, DeviceConfig(**cfg))

    # 创建 InfluxDB 写入器
    influx = AsyncInfluxWriter(
        url="http://localhost:8086",
        token="your-token",
        org="archive",
        bucket="env_monitor"
    )
    await influx.start()

    # 创建调度器
    scheduler = Scheduler(pool, influx)

    try:
        await scheduler.start()
    except KeyboardInterrupt:
        pass
    finally:
        await scheduler.stop()
        await influx.stop()

if __name__ == "__main__":
    asyncio.run(main())

# 方案 1：随机抖动
await asyncio.sleep(random.uniform(0, 5))

# 方案 2：按设备哈希均匀分布
offset = (hash(name) % 100) / 100 * poll_interval
await asyncio.sleep(offset)

# 方案 3：按区域分批
# 区域 A 设备：第 0-2 秒
# 区域 B 设备：第 2-4 秒
# 区域 C 设备：第 4-6 秒