106 lines
4.1 KiB
Python
106 lines
4.1 KiB
Python
|
# -*- coding: utf-8 -*-
|
||
|
|
"""本地 spool 缓存 + 断点续传 —— 丢失率 ≤ 0.02% 的实现保障（PRD 9 章）。
|
||
|
|
|
||
|
|
机制：
|
||
|
|
1. 每次采集样本先写入本地 spool 文件（JSON Lines，按小时分片）；
|
||
|
|
2. Kafka 上行确认（ack）后才删除对应记录；
|
||
|
|
3. 网关重启时扫描 spool 目录，未确认记录全部重发 —— 断点续传；
|
||
|
|
4. 上行通道抖动不丢数据，仅增加本地缓存占用（受 cache_limit_bytes 约束）。
|
||
|
|
|
||
|
|
文件命名：{shard_time:%Y%m%d%H}.spool.jsonl
|
||
|
|
"""
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import json
|
||
|
|
import os
|
||
|
|
import threading
|
||
|
|
import time
|
||
|
|
from typing import List
|
||
|
|
|
||
|
|
from drivers.base import SampleValue
|
||
|
|
|
||
|
|
|
||
|
|
class SpoolStore:
|
||
|
|
"""本地 spool：追加写 + 按 ack 删除（断点续传）。"""
|
||
|
|
|
||
|
|
def __init__(self, spool_dir: str, cache_limit_bytes: int = 512 * 1024 * 1024):
|
||
|
|
self.spool_dir = spool_dir
|
||
|
|
self.cache_limit_bytes = cache_limit_bytes
|
||
|
|
os.makedirs(spool_dir, exist_ok=True)
|
||
|
|
self._lock = threading.Lock()
|
||
|
|
|
||
|
|
# ------------------------------------------------------------------
|
||
|
|
def _shard_path(self, ts: float) -> str:
|
||
|
|
return os.path.join(
|
||
|
|
self.spool_dir,
|
||
|
|
time.strftime("%Y%m%d%H", time.localtime(ts)) + ".spool.jsonl",
|
||
|
|
)
|
||
|
|
|
||
|
|
def append(self, device_id: str, point_id: str, value: SampleValue, ts: float) -> None:
|
||
|
|
"""写入一条待上行的样本记录。"""
|
||
|
|
record = {
|
||
|
|
"device_id": device_id,
|
||
|
|
"point_id": point_id,
|
||
|
|
"value": value,
|
||
|
|
"ts": ts,
|
||
|
|
}
|
||
|
|
with self._lock:
|
||
|
|
with open(self._shard_path(ts), "a", encoding="utf-8") as fh:
|
||
|
|
fh.write(json.dumps(record, ensure_ascii=False) + "\n")
|
||
|
|
|
||
|
|
def pending_records(self, limit: int = 10000) -> List[dict]:
|
||
|
|
"""读取所有未确认记录（断点续传：重启后调用，全部重发）。"""
|
||
|
|
records: List[dict] = []
|
||
|
|
with self._lock:
|
||
|
|
for name in sorted(os.listdir(self.spool_dir)):
|
||
|
|
if not name.endswith(".spool.jsonl"):
|
||
|
|
continue
|
||
|
|
path = os.path.join(self.spool_dir, name)
|
||
|
|
with open(path, "r", encoding="utf-8") as fh:
|
||
|
|
for line in fh:
|
||
|
|
line = line.strip()
|
||
|
|
if line:
|
||
|
|
records.append(json.loads(line))
|
||
|
|
if len(records) >= limit:
|
||
|
|
return records
|
||
|
|
return records
|
||
|
|
|
||
|
|
def ack(self, record: dict) -> None:
|
||
|
|
"""上行确认后删除对应记录。
|
||
|
|
|
||
|
|
record 中 value/ts 为 None 的字段不参与匹配（Kafka 投递回调
|
||
|
|
仅有 point_id 时，按 point_id 删除最早一条未确认记录）。
|
||
|
|
"""
|
||
|
|
with self._lock:
|
||
|
|
for name in sorted(os.listdir(self.spool_dir)):
|
||
|
|
if not name.endswith(".spool.jsonl"):
|
||
|
|
continue
|
||
|
|
path = os.path.join(self.spool_dir, name)
|
||
|
|
try:
|
||
|
|
with open(path, "r", encoding="utf-8") as fh:
|
||
|
|
lines = fh.readlines()
|
||
|
|
except OSError:
|
||
|
|
continue
|
||
|
|
kept, removed = [], False
|
||
|
|
for line in lines:
|
||
|
|
line = line.strip()
|
||
|
|
if not line:
|
||
|
|
continue
|
||
|
|
parsed = json.loads(line)
|
||
|
|
matched = all(
|
||
|
|
record.get(k) is None or parsed.get(k) == v
|
||
|
|
for k, v in record.items()
|
||
|
|
)
|
||
|
|
if not removed and matched:
|
||
|
|
removed = True # 删除第一条匹配记录
|
||
|
|
else:
|
||
|
|
kept.append(line)
|
||
|
|
if removed:
|
||
|
|
with open(path, "w", encoding="utf-8") as fh:
|
||
|
|
fh.write("\n".join(kept) + ("\n" if kept else ""))
|
||
|
|
break
|
||
|
|
|
||
|
|
def total_pending(self) -> int:
|
||
|
|
"""当前未确认记录总数（健康度上报用）。"""
|
||
|
|
return len(self.pending_records(limit=10 ** 9))
|