Python代码片段-断点任务

使用Python处理一堆长耗时任务的时候，为了防止异常退出程序或者手动退出程序后丢失任务进度，可用使用断点的方式记录任务进度，下次重载任务后，继续运行上次未完成的任务即可。

这里用json文件作为数据持久化的方式，免去了使用数据库来作为持久的依赖问题，为了一个小的任务来搭建一套数据库，耗时耗力，得不偿失，不如选择本地文件来做简单的持久化，就算丢失部分任务进度，少数任务重新执行所消耗的时间完全处于容忍范围内。另外使用json文件记录任务进度，可用修改json文件来人为干预任务的执行情况，也是不错的一个优点

import json
import time
import osclass Task:def __init__(self, task_file='task_progress.json'):self.task_file = task_fileself.progress = []if os.path.exists(self.task_file):self.load_progress()def load_progress(self):"""加载任务进度"""with open(self.task_file, 'r') as f:data = json.load(f)self.progress = data.get('progress', 0)def save_progress(self):"""保存任务进度"""data = {'progress': self.progress}with open(self.task_file, 'w') as f:json.dump(data, f,indent=4, separators=(',', ': '))f.flush()os.fsync(f.fileno())  # 确保数据已写入磁盘def execute_task(self):legacy_tasks = [t  for t in self.progress if t['finish'] ==0 ]for i, v in enumerate(legacy_tasks):print(f"执行任务 {v['key']} ...")time.sleep(3)  # 模拟任务执行的时间self.mark_finished( v['key']) # 更行进度self.save_progress()  # 每次执行后保存进度print(f"任务 {v['key']} 完成。")print("任务完成！")def flush_progresses(self):tasks = [{"key":"xxx1"},{"key":"xxx2"},{"key":"xxx3"}]self.progress = [ {"key" : t['key'], 'finish':0} for t in tasks]self.save_progress()def mark_finished(self, key):"""标记完成"""print(key)for i, t in enumerate(self.progress) :if t ['key'] ==  key :self.progress[i]['finish'] = 1breakself.save_progress()def count(self):"""查询未完成"""return len( [t for t in self.progress if t['finish'] == 0])if __name__ == "__main__":task = Task()# 第一次生成任务列表，下次任务时候注释掉此行task.flush_progresses()try:task.execute_task()except KeyboardInterrupt:print("\n任务被中断，进度已保存。")