运行Python脚本怎样恢复中断的脚本执行 运行Python脚本的断点续执行技巧

import signal
import time
import os

STATE_FILE = "script_state.txt"
interrupted = False

def signal_handler(sig, frame):
    global interrupted
    interrupted = True
    print("脚本中断，正在保存状态...")

signal.signal(signal.SIGINT, signal_handler)

def save_state(counter):
    with open(STATE_FILE, "w") as f:
        f.write(str(counter))

def load_state():
    if os.path.exists(STATE_FILE):
        with open(STATE_FILE, "r") as f:
            return int(f.read())
    return 0

start_counter = load_state()
print(f"从计数器 {start_counter} 开始...")

for i in range(start_counter, 100):
    if interrupted:
        save_state(i)
        print("状态已保存，脚本退出。")
        break
    print(f"计数: {i}")
    time.sleep(1)
else:
    if os.path.exists(STATE_FILE):
        os.remove(STATE_FILE) # 任务完成，删除状态文件
    print("任务完成!")

script_state.txt

try...except

try:
    # 可能引发异常的代码
    result = 10 / 0
except ZeroDivisionError as e:
    # 处理 ZeroDivisionError 异常
    print(f"发生除零错误: {e}")
except Exception as e:
    # 处理其他类型的异常
    print(f"发生未知错误: {e}")
else:
    # 如果没有发生异常，则执行此代码块
    print(f"结果是: {result}")
finally:
    # 无论是否发生异常，都会执行此代码块
    print("程序执行完毕")

else

try

finally

import logging

logging.basicConfig(filename='error.log', level=logging.ERROR)

try:
    result = 10 / 0
except ZeroDivisionError as e:
    logging.error(f"发生除零错误: {e}")
except Exception as e:
    logging.exception("发生未知错误") # 记录完整的堆栈信息

class MyCustomError(Exception):
    def __init__(self, message):
        super().__init__(message)
        self.message = message

try:
    raise MyCustomError("自定义错误")
except MyCustomError as e:
    print(f"发生自定义错误: {e.message}")

retry

from retry import retry

@retry(tries=3, delay=2, backoff=2)
def unreliable_function():
    print("尝试执行...")
    # 模拟一个可能失败的操作
    import random
    if random.random() < 0.5:
        raise IOError("操作失败")
    print("操作成功!")

unreliable_function()

tries

delay

backoff

from retry import retry

def is_temporary_error(exception):
    return isinstance(exception, IOError)

@retry(retry_on_exception=is_temporary_error, tries=3, delay=2)
def unreliable_function():
    print("尝试执行...")
    # 模拟一个可能失败的操作
    import random
    if random.random() < 0.5:
        raise IOError("操作失败")
    print("操作成功!")

unreliable_function()

multiprocessing

import multiprocessing
import time

def worker(num):
    print(f"进程 {num} 开始...")
    time.sleep(2)
    print(f"进程 {num} 结束.")

if __name__ == '__main__':
    processes = []
    for i in range(3):
        p = multiprocessing.Process(target=worker, args=(i,))
        processes.append(p)
        p.start()

    for p in processes:
        p.join()

    print("所有进程结束.")

threading

import threading
import time

def worker(num):
    print(f"线程 {num} 开始...")
    time.sleep(2)
    print(f"线程 {num} 结束.")

threads = []
for i in range(3):
    t = threading.Thread(target=worker, args=(i,))
    threads.append(t)
    t.start()

for t in threads:
    t.join()

print("所有线程结束.")

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor
import time

def worker(num):
    print(f"任务 {num} 开始...")
    time.sleep(2)
    print(f"任务 {num} 结束.")
    return num * 2

with ThreadPoolExecutor(max_workers=3) as executor:
    results = executor.map(worker, range(5))
    for result in results:
        print(f"结果: {result}")

with ProcessPoolExecutor(max_workers=3) as executor:
    results = executor.map(worker, range(5))
    for result in results:
        print(f"结果: {result}")