Django 如何并发查询远程噪声计并实时更新浏览器数据？

建议将噪声计采集程序写成独立运行的 Django management command。每台设备对应一个长期运行的 asyncio 任务，主动建立 TCP 连接、定时查询，并通过 Channels 的 group_send() 发布数据。现有 WebSocket Consumer 负责订阅分组，将数据转发给浏览器。

多位用户查看同一台噪声计时，可以共享一份采集结果；不同噪声计的采集任务则通过异步 I/O 并发运行。

先修改原有查询代码

原代码中最影响并发的是 time.sleep()。即使写在 async def 中，它仍会阻塞当前线程，让同一事件循环中的其他设备任务无法继续运行。应将这些调用全部替换为：

await asyncio.sleep(1)

TCP 响应的边界也需要处理。reader.read(1024) 最多读取 1024 字节，不能保证恰好读到一条完整响应。具体读取方式取决于仪器协议：

• 响应以换行符结束时，使用 readuntil(b"\n")。
• 响应长度固定时，使用 readexactly()。
• 使用长度字段或其他结束标记时，按协议解析。

同一台设备的命令仍应按顺序执行。共用一个 TCP 连接时，不要让多个协程同时读取同一个 StreamReader，否则无法可靠地匹配命令和响应。

用独立进程运行采集任务

建议按以下流程传递数据：

多台噪声计
    ↓ TCP
采集进程：每台设备一个 asyncio 任务
    ↓ group_send()
共享 Channel Layer
    ↓
WebSocket Consumer
    ↓
浏览器

采集进程可以使用 Django 的配置和业务代码，生命周期独立于 HTTP 请求和 WebSocket 连接。浏览器断开后，采集仍可继续运行。

不要在模块导入时或 AppConfig.ready() 中直接启动采集任务。开发环境的自动重载和多个 ASGI worker 都可能重复启动任务，导致多个采集任务同时操作同一台设备。

配置跨进程共享的 Channel Layer

采集命令和 ASGI 服务运行在不同进程中时，需要连接同一个共享 Channel Layer。使用 channels_redis 时可按下面配置，并将连接地址替换为实际地址：

# settings.py
CHANNEL_LAYERS = {
    "default": {
        "BACKEND": "channels_redis.core.RedisChannelLayer",
        "CONFIG": {
            "hosts": [("127.0.0.1", 6379)],
        },
    },
}

InMemoryChannelLayer 的数据只存在于当前进程，无法将独立采集进程的消息传给另一个进程中的 Consumer。

为每台噪声计创建一个异步任务

在已注册的 Django 应用中，可以将示例代码放到以下路径：

meters/management/commands/collect_meters.py

运行示例需要满足以下前提：

• 仪器响应以 \n 结束，内容可以按 ASCII 解码。
• INIT START、MEAS:INIT、INIT STOP 不返回确认消息，查询命令返回一条响应。
• 仪器已完成所需配置，允许按示例顺序启动测量、触发和查询。
• 查询返回的是仪器协议定义的数据，不保证一定是刚触发的测量结果。

如果设备的实际行为不同，需要先调整命令顺序、等待条件和响应读取方式。代码中的超时、启动等待和重连间隔均为示例值。

import asyncio
import logging
from datetime import datetime, timezone

from channels.layers import get_channel_layer
from django.core.management.base import BaseCommand


logger = logging.getLogger(__name__)

# 示例配置；实际项目可从数据库或 Django settings 读取。
METERS = [
    {"id": "1", "host": "192.168.10.1", "port": 6889},
    {"id": "2", "host": "192.168.10.2", "port": 6889},
]

INTERVAL = 1.0
IO_TIMEOUT = 3.0
RECONNECT_DELAY = 3.0


async def send_command(writer, command):
    writer.write(command.encode("ascii"))
    await asyncio.wait_for(writer.drain(), timeout=IO_TIMEOUT)


async def query(reader, writer, command):
    await send_command(writer, command)
    data = await asyncio.wait_for(
        reader.readuntil(b"\n"),
        timeout=IO_TIMEOUT,
    )
    return data.decode("ascii").rstrip("\r\n")


async def collect_session(meter, channel_layer):
    reader, writer = await asyncio.wait_for(
        asyncio.open_connection(meter["host"], meter["port"]),
        timeout=IO_TIMEOUT,
    )

    try:
        # 按设备手册补充身份查询、参数配置及必要的确认响应读取。
        await send_command(writer, "INIT START\n")
        await asyncio.sleep(6)

        loop = asyncio.get_running_loop()

        while True:
            started_at = loop.time()

            await send_command(writer, "MEAS:INIT\n")
            value = await query(
                reader,
                writer,
                "MEAS:SLM:123? LAEQ\n",
            )

            payload = {
                "meter_id": meter["id"],
                "received_at": datetime.now(timezone.utc).isoformat(),
                "value": value,
            }

            try:
                await asyncio.wait_for(
                    channel_layer.group_send(
                        f"meter.{meter['id']}",
                        {
                            "type": "meter.reading",
                            "payload": payload,
                        },
                    ),
                    timeout=IO_TIMEOUT,
                )
            except Exception:
                # 发布失败时记录本次丢失，继续维持设备采集连接。
                logger.exception(
                    "Failed to publish reading for meter %s",
                    meter["id"],
                )

            elapsed = loop.time() - started_at
            await asyncio.sleep(max(0.0, INTERVAL - elapsed))

    finally:
        try:
            await send_command(writer, "INIT STOP\n")
        except (OSError, asyncio.TimeoutError):
            pass
        finally:
            writer.close()
            try:
                await asyncio.wait_for(
                    writer.wait_closed(),
                    timeout=IO_TIMEOUT,
                )
            except (OSError, asyncio.TimeoutError):
                pass


async def collect_forever(meter, channel_layer):
    while True:
        try:
            await collect_session(meter, channel_layer)
        except asyncio.CancelledError:
            raise
        except (
            OSError,
            EOFError,
            asyncio.TimeoutError,
            asyncio.LimitOverrunError,
            UnicodeError,
        ):
            logger.exception(
                "Connection or protocol failure for meter %s",
                meter["id"],
            )
            await asyncio.sleep(RECONNECT_DELAY)


async def run_collectors():
    channel_layer = get_channel_layer()
    if channel_layer is None:
        raise RuntimeError("CHANNEL_LAYERS is not configured")

    tasks = [
        asyncio.create_task(collect_forever(meter, channel_layer))
        for meter in METERS
    ]

    try:
        await asyncio.gather(*tasks)
    finally:
        for task in tasks:
            task.cancel()
        await asyncio.gather(*tasks, return_exceptions=True)


class Command(BaseCommand):
    help = "Collect noise meter readings and publish them through Channels"

    def handle(self, *args, **options):
        try:
            asyncio.run(run_collectors())
        except KeyboardInterrupt:
            pass

启动采集进程：

python manage.py collect_meters

management command 是同步入口，因此这里用 asyncio.run() 创建事件循环。在已有的异步 Consumer 中，应直接 await 异步操作，不能嵌套调用 asyncio.run()。

示例省略了原代码中的 *RST 和自动保存按键序列。首次连接或重连时是否执行这些命令，需要根据仪器状态和测量流程决定，尤其要确认复位是否会中断正在进行的测量。

将采集结果转发给浏览器

每台设备使用一个分组，例如 meter.1。浏览器订阅设备后，Consumer 加入对应分组。

在现有 Consumer 的 connect() 中，完成身份认证和设备访问权限检查后，加入以下代码：

self.meter_group = f"meter.{meter_id}"

await self.channel_layer.group_add(
    self.meter_group,
    self.channel_name,
)
await self.accept()

meter_id 应是经过校验的设备标识。示例使用数字 ID，避免将含有特殊字符的设备名称直接拼入分组名。

如果现有 Consumer 继承自 AsyncJsonWebsocketConsumer，增加以下事件处理方法：

async def meter_reading(self, event):
    await self.send_json(event["payload"])

group_send() 中的 "type": "meter.reading" 对应 Consumer 的 meter_reading() 方法。

断开连接时退出分组：

async def disconnect(self, close_code):
    if hasattr(self, "meter_group"):
        await self.channel_layer.group_discard(
            self.meter_group,
            self.channel_name,
        )

浏览器收到的消息形式如下：

{
  "meter_id": "1",
  "received_at": "2026-01-01T08:00:00+00:00",
  "value": "仪器返回的原始响应"
}

沿用现有 WebSocket 消息处理逻辑，根据 meter_id 更新对应显示区域即可。received_at 是服务器收到响应后的时间；如果需要仪器的实际测量时间，应从设备响应中读取。

查询频率和部署时需要注意什么

每秒查询取决于设备响应速度

如果直接在每轮末尾执行 await asyncio.sleep(1)，实际周期就是查询耗时再加一秒。示例扣除了本轮耗时，在查询和发布总耗时小于一秒时，尝试维持约一秒的查询周期。

如果一次查询已经超过一秒，下一轮只能等本轮结束后再开始。不要在同一连接上叠加查询来追赶频率。如果需要严格的采样周期，应确认仪器是否支持内部定时采样、缓存读取或主动推送。

每台设备应有唯一的采集任务

增加 ASGI worker 数量时，无需相应增加采集任务。部署时先运行一个采集进程；需要扩展采集能力时，再按设备分配任务，确保同一台设备只由一个采集进程管理。跨进程的唯一性需要通过任务分配或分布式锁等方式保证，普通的 asyncio.Lock 无法覆盖多个进程。

各设备通过独立的异步任务并发采集：一台设备等待网络响应时，事件循环可以处理其他设备。某台设备超时后，只会进入自己的重连循环。

分组广播不保存历史数据

Channel Layer 用于传递实时事件，不能保证每次测量结果都能到达浏览器。示例在发布异常时记录日志并继续采集，因此该次数据可能丢失。

如果需要在刷新页面后立即显示最新值，可以将最新结果另存到共享缓存；如果需要完整的测量记录，应独立持久化。浏览器还应根据最后更新时间显示数据是否过期，避免设备断线后仍将旧值显示为当前读数。

备注：内容仅供参考。