长时任务与自适应优化

New-Item -ItemType Directory -Force l9-long-tasks
cd l9-long-tasks

mkdir -p l9-long-tasks
cd l9-long-tasks

import json
from pathlib import Path
import tempfile

BATCH = 2
ITEMS = ["订单A", "订单B", "订单C", "订单D"]

def advance(path, items, batch, fail_on=None):
    if batch < 1:
        raise ValueError("每批至少处理一项")
    # 检查点含已完成ID与下一位置；每项成功后才推进位置。
    state = json.loads(path.read_text(encoding="utf-8")) if path.exists() else {"next": 0, "done": []}
    next_index = state.get("next")
    # bool在Python中也是int的子类，显式type检查可拒绝True这样的损坏位置。
    if type(next_index) is not int or not 0 <= next_index <= len(items):
        raise ValueError("检查点位置必须为合法整数")
    if state.get("done") != items[:next_index]:
        raise ValueError("任务清单与检查点不一致")
    stop = min(state["next"] + batch, len(items))
    for index in range(state["next"], stop):
        if items[index] == fail_on:
            raise RuntimeError("当前项失败，检查点保留")
        state["done"].append(items[index])
        state["next"] = index + 1
        pending = path.with_suffix(".pending")
        pending.write_text(json.dumps(state, ensure_ascii=False), encoding="utf-8")
        pending.replace(path)
    return state

def adapt_batch(previous, failures, attempts, average_ms, remaining_ms):
    if previous < 1 or attempts < 1 or not 0 <= failures <= attempts:
        raise ValueError("批次与反馈数量无效")
    if average_ms <= 0 or remaining_ms < 0:
        raise ValueError("平均耗时需为正，剩余预算不能为负")
    # 失败比例达到25%先减半；低于25%则翻倍；再用剩余时间预算限制批次。
    affordable = int(remaining_ms // average_ms)
    candidate = max(1, previous // 2) if failures / attempts >= 0.25 else previous * 2
    return min(candidate, affordable)


if __name__ == "__main__":
    with tempfile.TemporaryDirectory() as folder:
        path = Path(folder) / "checkpoint.json"
        first = advance(path, ITEMS, BATCH)
        print("首次完成：" + ",".join(first["done"]))
        resumed = advance(path, ITEMS, BATCH)
        print("恢复完成：" + ",".join(resumed["done"]))
        print("下次位置：" + str(resumed["next"]))
    first_batch = adapt_batch(4, 1, 4, 200, 1200)
    second_batch = adapt_batch(first_batch, 0, 2, 200, 600)
    print("反馈调度1：失败1/4，均耗200ms，预算1200ms → 批量" + str(first_batch))
    print("反馈调度2：失败0/2，均耗200ms，预算600ms → 批量" + str(second_batch))

import unittest
from pathlib import Path
import tempfile
import checkpoint_lab as lab


class MechanismTests(unittest.TestCase):
    # 正常、边界和失败输入都由这里提供，直接运行即可。

    def test_normal_resume(self):
        with tempfile.TemporaryDirectory() as folder:
            path = Path(folder) / "state.json"
            lab.advance(path, lab.ITEMS, 2)
            self.assertEqual(lab.advance(path, lab.ITEMS, 2)["done"], lab.ITEMS)
        self.assertEqual(lab.adapt_batch(4, 1, 4, 200, 1200), 2)
        self.assertEqual(lab.adapt_batch(2, 0, 2, 200, 600), 3)

    def test_boundary_empty(self):
        with tempfile.TemporaryDirectory() as folder:
            self.assertEqual(lab.advance(Path(folder) / "s.json", [], 1)["next"], 0)
        self.assertEqual(lab.adapt_batch(1, 0, 1, 200, 199), 0)

    def test_failure_keeps_success(self):
        with tempfile.TemporaryDirectory() as folder:
            path = Path(folder) / "s.json"
            with self.assertRaises(RuntimeError):
                lab.advance(path, lab.ITEMS, 4, "订单B")
            self.assertEqual(lab.advance(path, lab.ITEMS, 1)["done"], ["订单A", "订单B"])
            with self.assertRaises(ValueError):
                lab.advance(path, lab.ITEMS, 0)
            # 损坏的磁盘材料：负数、布尔、非整数、越界都在切片前拒绝。
            for position in (-1, True, 1.5, 5):
                path.write_text(lab.json.dumps({"next": position, "done": []}), encoding="utf-8")
                with self.assertRaises(ValueError):
                    lab.advance(path, lab.ITEMS, 1)


if __name__ == "__main__":
    unittest.main()

# 在刚才创建的练习目录检查文件，再运行程序。
ls
python -X utf8 checkpoint_lab.py
python -X utf8 -m unittest -v test_checkpoint_lab.py

首次完成：订单A,订单B
恢复完成：订单A,订单B,订单C,订单D
下次位置：4
反馈调度1：失败1/4，均耗200ms，预算1200ms → 批量2
反馈调度2：失败0/2，均耗200ms，预算600ms → 批量3

python -X utf8 checkpoint_lab.py