综合实时python案例，中场休息会如何调整？

综合实时Python案例：中场休息调整策略

"中场休息调整"在实时系统中通常指运行过程中的动态干预/热更新/状态切换，下面用几个真实场景的Python案例来说明。

案例1：实时数据采集系统 —— 动态降频/升频

场景：传感器每100ms采样一次，中场休息（如午间）自动降频省资源，恢复时升频。

import time
import threading
from datetime import datetime
class AdaptiveSampler:
    def __init__(self):
        self.interval = 0.1          # 正常采样间隔
        self.running = True
        self.mode = "normal"          # normal / rest
        self.lock = threading.Lock()
    def set_rest_mode(self, enable: bool):
        """中场休息切换"""
        with self.lock:
            if enable:
                self.mode = "rest"
                self.interval = 1.0   # 降频10倍
                print(f"[{datetime.now():%H:%M:%S}] 🌙 进入休息模式，采样间隔→1s")
            else:
                self.mode = "normal"
                self.interval = 0.1
                print(f"[{datetime.now():%H:%M:%S}] ☀️  恢复工作模式，采样间隔→0.1s")
    def run(self):
        while self.running:
            with self.lock:
                interval = self.interval
            # 模拟采集
            print(f"采集数据 @ {time.time():.2f} (mode={self.mode})")
            time.sleep(interval)
# 使用：主线程里定时切换
sampler = AdaptiveSampler()
t = threading.Thread(target=sampler.run, daemon=True)
t.start()
time.sleep(3)
sampler.set_rest_mode(True)   # 中场休息
time.sleep(5)
sampler.set_rest_mode(False)  # 恢复
time.sleep(2)
sampler.running = False

要点：用 threading.Lock 保护共享状态，主循环每次读取最新 interval，实现无停机调整。

案例2：实时交易/行情推送 —— 暂停订阅 + 状态快照

场景：WebSocket 实时行情，中场休息时暂停推送、保存断点，恢复时从快照续传。

import asyncio, json, time
class MarketStream:
    def __init__(self):
        self.paused = asyncio.Event()
        self.paused.set()                # 初始 = 运行
        self.last_snapshot = None
        self.seq = 0
    async def produce(self):
        while True:
            await self.paused.wait()     # 休息时挂起
            self.seq += 1
            msg = {"seq": self.seq, "price": 100 + self.seq % 5, "ts": time.time()}
            self.last_snapshot = msg
            print("推送:", msg)
            await asyncio.sleep(0.5)
    async def take_break(self):
        print("⏸  中场休息：暂停推送，保存快照")
        self.paused.clear()              # 阻塞 produce
        await asyncio.sleep(2)
        print(f"▶️  恢复推送，从 seq={self.last_snapshot['seq']} 续传")
        self.paused.set()
async def main():
    stream = MarketStream()
    task = asyncio.create_task(stream.produce())
    await asyncio.sleep(2)
    await stream.take_break()
    await asyncio.sleep(2)
    task.cancel()
asyncio.run(main())

要点：asyncio.Event 做背压/暂停开关，快照保证恢复后数据不丢。

案例3：实时计算任务 —— 检查点 + 优雅暂停

场景：长时间实时聚合任务，中场休息时 checkpoint 到磁盘，进程可安全重启。

import signal, json, time, os
class ResumableAggregator:
    CKPT = "agg_ckpt.json"
    def __init__(self):
        self.total = 0
        self.count = 0
        self.running = True
        self._load()
    def _load(self):
        if os.path.exists(self.CKPT):
            with open(self.CKPT) as f:
                d = json.load(f)
            self.total, self.count = d["total"], d["count"]
            print(f"♻️  从检查点恢复: total={self.total}, count={self.count}")
    def _save(self):
        with open(self.CKPT, "w") as f:
            json.dump({"total": self.total, "count": self.count}, f)
        print("💾 检查点已保存")
    def handle_pause(self, signum, frame):
        print("收到中场休息信号，优雅停止中...")
        self.running = False
    def run(self):
        signal.signal(signal.SIGINT, self.handle_pause)
        while self.running:
            self.total += 1          # 模拟业务
            self.count += 1
            if self.count % 10 == 0:
                self._save()
            time.sleep(0.2)
        self._save()
        print("✅ 已安全暂停，可下次恢复")
ResumableAggregator().run()

要点：signal 捕获中断 → 落盘 → 退出；重启时 _load 续算，实现中断可恢复。

案例4：实时监控告警 —— 抑制规则（休息期不报警）

场景：服务监控，中场休息（如维护窗口）时抑制告警，但仍记录日志。

import time
from contextlib import contextmanager
class AlertManager:
    def __init__(self):
        self.suppressed = False
        self.buffer = []
    @contextmanager
    def maintenance_window(self):
        """中场休息：进入维护窗口，抑制告警"""
        self.suppressed = True
        print("🔕 进入维护窗口，告警抑制中")
        try:
            yield
        finally:
            self.suppressed = False
            print(f"🔔 退出维护窗口，休息期间缓存 {len(self.buffer)} 条事件")
            for e in self.buffer:
                self._send(e)
            self.buffer.clear()
    def alert(self, msg):
        if self.suppressed:
            self.buffer.append(msg)       # 缓冲不推送
            print(f"  [缓冲] {msg}")
        else:
            self._send(msg)
    def _send(self, msg):
        print(f"  🚨 发送告警: {msg}")
# 使用
am = AlertManager()
am.alert("CPU 80%")
with am.maintenance_window():        # 中场休息
    time.sleep(1)
    am.alert("CPU 95%")              # 被抑制
    am.alert("内存 90%")             # 被抑制
am.alert("磁盘 85%")                 # 正常发出

要点：上下文管理器 + 缓冲区，休息期静默、恢复后补发。

中场休息调整的通用设计套路

核心原则：可中断、可恢复、可观测（保存 checkpoint + 保持状态一致 + 日志明确边界）。