一、 场景痛点：内存队列的“爆栈”噩梦

周末收到一个做智慧矿山项目的开发团队求助：

• 现状：他们使用 Python (queue.Queue) 在边缘网关上做了一个内存队列，缓存从 PLC 采集的数据，然后通过 MQTT 发往云端。

• 故障：矿山现场的 4G 信号极不稳定，经常断网 1-2 小时。

• 后果：

1. OOM (Out of Memory)：断网期间，采集线程还在疯狂塞数据，内存队列无限制增长，最终耗尽网关的 4GB 内存，系统强制杀进程（OOM Killer）。

2. 数据丢失：进程重启后，内存里缓存的 1 小时关键数据全部丢失，甲方无法接受。

我们需要引入一个“持久化缓冲层”，它既要比 SQL 数据库快，又要能抵抗断网带来的背压。2026 年的最佳实践是：Redis Stream。

二、 架构设计：生产者-消费者解耦

我们将单体脚本拆分为两个独立的进程，中间通过 Redis Stream 解耦：

1. 生产者 (Collector)：只负责从 PLC 读数据，并 XADD 写入 Redis。它不关心网络好不好，只关心写入速度（极快）。

2. 缓冲层 (Broker)：Redis Stream，设置最大长度（MAXLEN），防止磁盘被写满。

3. 消费者 (Uploader)：负责从 Redis 读取数据发往云端。如果断网，它就阻塞重试；如果联网，它就全力消费。

PLC -> [采集进程] --(XADD)--> [Redis Stream (持久化/自动丢弃旧数据)] --(XREADGROUP)--> [上传进程] -> MQTT Broker

三、 核心实施步骤 (Copy & Paste)

假设网关已安装 Docker，拉取 Redis 镜像：docker run -d -p 6379:6379 --name edge-redis redis:7.4 --appendonly yes

1. 生产者：带“熔断机制”的写入

在采集脚本中，使用 MAXLEN 限制队列长度。例如限制缓存 10 万条（约 20MB），超过自动剔除最旧数据，保命优先。

import redis
import json
import time

r = redis.Redis(host='localhost', port=6379, decode_responses=True)
STREAM_KEY = "sensor_events"

def collect_data():
    while True:
        data = {"ts": time.time(), "val": 24.5} # 模拟 PLC 数据
        
        # 核心代码：XADD
        # maxlen=100000: 约等于缓存 2 小时数据
        # approximate=True (~): 提升性能，不强求精确剔除
        try:
            r.xadd(STREAM_KEY, data, maxlen=100000, approximate=True)
        except redis.exceptions.ConnectionError:
            print("Redis 连接失败，启用本地文件降级策略...")
        
        time.sleep(0.1) # 10Hz 采样

2. 消费者：支持“断点续传”的读取

上传脚本使用消费者组 (Consumer Group) 模式，确保数据不丢、不重。

Python

# 初始化消费者组 (仅运行一次)
try:
    r.xgroup_create(STREAM_KEY, "cloud_uploader_group", id="0", mkstream=True)
except redis.exceptions.ResponseError:
    pass # 组已存在

def upload_worker():
    while True:
        # 1. 阻塞读取数据 (Block 2秒)
        # '>' 表示读取本组尚未消费的最新消息
        entries = r.xreadgroup("cloud_uploader_group", "worker_1", {STREAM_KEY: ">"}, count=50, block=2000)
        
        if not entries:
            continue # 没数据，或者网络好都发完了
            
        stream, messages = entries[0]
        
        for msg_id, msg_body in messages:
            # 2. 发送 MQTT (模拟)
            if send_to_cloud(msg_body):
                # 3. 关键：确认消费 (ACK)
                # 只有 ACK 了，Redis 才会认为这条处理完了，否则下次重启还会发
                r.xack(STREAM_KEY, "cloud_uploader_group", msg_id)
            else:
                # 发送失败 (断网)，跳出循环，等待下一轮重试
                # 此时 msg_id 仍在 Pending 列表里，不会丢失
                time.sleep(5) 
                break

四、 踩坑复盘 (Red Flags)

1. Flash 寿命损耗

• 风险：Redis 默认的 AOF (Append Only File) 是每秒刷盘 (appendfsync everysec)。如果你的工业网关用的是普通的 TF 卡，高频写入（100Hz）会在半年内把 TF 卡写废。

• 对策：

• 硬件上：务必使用板载 eMMC 或 NVMe SSD。

• 配置上：如果对数据丢失容忍度稍高，可将 AOF 设置为 no（由操作系统决定刷盘），或者购买带 超级电容 的网关。

2. 消费者“假死” (PEL 堆积)

• 现象：程序重启后，有些数据永远发不上去了。

• 原因：代码逻辑 Bug 导致 send_to_cloud 失败但程序没崩溃，数据一直处于“已读取未 ACK”状态（Pending）。

• 对策：必须实现一个 "Pending 扫描器"。启动时先检查 XPENDING 列表，把那些处理超时的消息抢救回来重新发。

3. 消息体过大

• 建议：Redis Stream 适合存元数据或小 Payload。如果你要传图片或波形文件，不要直接塞进 Stream。应该把文件存文件系统，Stream 里只存文件路径。

五、 关联资源与选型

这套架构需要网关具备一定的内存和 IO 性能。

• 硬件推荐：

• 研华 UNO-137 (Intel Atom)：支持 NVMe 扩展，完美承载 Redis 高频 IO。

  
  

• 树莓派 5 (8GB) + NVMe Base：消费级中最强的 IO 组合，适合非严苛工业环境。

  
  

一键部署

这是一个标准化的“边缘缓冲微服务”。

我们提供了 docker-compose.yml，内含优化过配置的 Redis 7.4 和 Python 生产者/消费者 Demo。