4. 边缘数据缓存与断点续传

任务说明

在边缘计算场景中，网络连接往往不稳定。本案例实现一个最小可用的边缘数据缓存与断点续传机制：传感器数据持续产生，当网络不可用时数据暂存于本地缓存文件；网络恢复后，自动将缓存数据按序补传到云端，并清理已成功上传的缓存。

实现方案

使用Python标准库实现，不依赖第三方包。核心思路：

• 数据产生端模拟传感器，每0.5秒产生一条带时间戳的数据；
• 网络状态通过一个可手动切换的开关模拟（真实场景可替换为HTTP请求探测）；
• 缓存采用本地文件追加写入，每条记录一行JSON；
• 断点续传通过记录文件读取偏移量实现，避免重复读取已上传数据。

完整代码

import json
import os
import time
from datetime import datetime

CACHE_FILE = "edge_cache.jsonl"
OFFSET_FILE = "cache_offset.txt"
UPLOADED_FILE = "uploaded_records.jsonl"

# ---------- 网络状态模拟 ----------
class NetworkSimulator:
    def __init__(self):
        self.online = True

    def is_connected(self):
        return self.online

    def set_online(self, status):
        self.online = status

net = NetworkSimulator()

# ---------- 缓存写入 ----------
def cache_data(record):
    with open(CACHE_FILE, "a", encoding="utf-8") as f:
        f.write(json.dumps(record, ensure_ascii=False) + "\n")

# ---------- 断点续传 ----------
def get_offset():
    if os.path.exists(OFFSET_FILE):
        with open(OFFSET_FILE, "r") as f:
            return int(f.read().strip())
    return 0

def save_offset(pos):
    with open(OFFSET_FILE, "w") as f:
        f.write(str(pos))

def upload_pending():
    if not os.path.exists(CACHE_FILE):
        return 0

    offset = get_offset()
    uploaded_count = 0

    with open(CACHE_FILE, "r", encoding="utf-8") as f:
        f.seek(offset)
        for line in f:
            line = line.strip()
            if not line:
                continue
            record = json.loads(line)
            # 模拟上传到云端
            with open(UPLOADED_FILE, "a", encoding="utf-8") as out:
                out.write(json.dumps(record, ensure_ascii=False) + "\n")
            uploaded_count += 1
            offset = f.tell()  # 更新为当前读取位置

    save_offset(offset)
    return uploaded_count

# ---------- 主循环 ----------
def run(duration_seconds=10):
    start = time.time()
    seq = 0

    while time.time() - start < duration_seconds:
        # 模拟网络波动：每3秒切换一次状态
        if int(time.time()) % 6 < 3:
            net.set_online(True)
        else:
            net.set_online(False)

        record = {
            "seq": seq,
            "timestamp": datetime.now().isoformat(),
            "value": round(seq * 0.5, 2),
        }
        seq += 1

        if net.is_connected():
            # 网络在线：先尝试补传缓存，再直接上传当前数据
            n = upload_pending()
            if n > 0:
                print(f"[上传] 补传缓存 {n} 条")
            with open(UPLOADED_FILE, "a", encoding="utf-8") as f:
                f.write(json.dumps(record, ensure_ascii=False) + "\n")
            print(f"[在线] 直接上传: seq={record['seq']}")
        else:
            # 网络离线：写入缓存
            cache_data(record)
            print(f"[离线] 已缓存: seq={record['seq']}")

        time.sleep(0.5)

    # 结束时若网络恢复，再尝试一次补传
    if net.is_connected():
        n = upload_pending()
        if n > 0:
            print(f"[结束补传] 补传 {n} 条")

if __name__ == "__main__":
    # 清理旧数据（便于重复运行验证）
    for f in [CACHE_FILE, OFFSET_FILE, UPLOADED_FILE]:
        if os.path.exists(f):
            os.remove(f)

    print("=== 边缘缓存与断点续传演示 ===")
    run(duration_seconds=10)

    # 输出验证统计
    cache_count = sum(1 for _ in open(CACHE_FILE, encoding="utf-8")) if os.path.exists(CACHE_FILE) else 0
    upload_count = sum(1 for _ in open(UPLOADED_FILE, encoding="utf-8")) if os.path.exists(UPLOADED_FILE) else 0
    print(f"\n=== 验证结果 ===")
    print(f"缓存文件剩余记录数: {cache_count}")
    print(f"已上传记录总数: {upload_count}")
    print(f"缓存偏移量: {get_offset()}")

运行方式

python edge_cache_demo.py

验证方法与通过条件

• 运行观察：程序运行10秒，控制台交替打印[在线]与[离线]状态，离线期间数据被缓存，在线时先补传缓存再上传新数据。
• 文件检查：运行结束后，打开uploaded_records.jsonl，其中应包含所有seq编号从0开始的连续记录（无缺失）；edge_cache.jsonl中可能残留最后离线阶段未来得及补传的数据（取决于结束时机），但cache_offset.txt指向的位置之前的数据均已成功上传。
• 判定标准：若uploaded_records.jsonl中记录条数 + edge_cache.jsonl中剩余条数 = 总产生条数（约20条），且seq连续无重复，则断点续传机制验证通过。

关键说明

• 网络状态每3秒切换一次，便于观察缓存与续传行为；实际部署时可将net.is_connected()替换为真实的云端心跳检测或HTTP请求。
• 断点续传依赖cache_offset.txt记录文件读取位置，即使程序中途崩溃，重启后也能从上次位置继续补传，不会重复上传。
• 本实现为单文件最小闭环，未引入多线程或异步框架；若需并发处理，可在upload_pending()基础上增加线程锁。