Python多线程场景下如何在信号或异常触发时保存set集合数据

核心优化点 #

• 补充线程锁保证多线程写入set的操作安全，避免数据错乱或丢失
• 注册系统信号处理器，覆盖SIGTERM、SIGINT等常见进程终止信号
• 调整异常捕获范围，覆盖普通异常、键盘中断、系统退出等所有终止场景
• 抽取公共数据保存函数，避免重复代码逻辑

完整实现代码 #

import logging
import signal
import threading
import requests
from concurrent.futures import ThreadPoolExecutor

# 初始化线程锁，保证set操作线程安全
lock = threading.Lock()
# 待校验数据集合，用户自行填充数据
s = set()
valid_data = set()

# 日志配置
logging.basicConfig(level=logging.INFO,
                    format='%(asctime)s - %(message)s',
                    datefmt='%Y-%m-%d %H:%M:%S',
                    handlers=[
                        logging.FileHandler("debug.log"),
                        logging.StreamHandler()
                    ]
                    )

# 公共保存数据函数
def save_data(file_path: str):
    with open(file_path, 'w', encoding='utf-8') as f:
        for d in valid_data:
            f.write(f"{d}\n")
    logging.info(f"数据已保存到{file_path}，共{len(valid_data)}条有效数据")

# 信号处理函数
def handle_signal(signum, frame):
    logging.warning(f"收到终止信号{signum}，即将保存数据并退出")
    # 触发异常让主线程执行流走到异常处理分支
    raise KeyboardInterrupt

# 注册需要监听的信号
signal.signal(signal.SIGINT, handle_signal)
signal.signal(signal.SIGTERM, handle_signal)
# Windows系统可额外注册SIGBREAK信号
# signal.signal(signal.SIGBREAK, handle_signal)

def data_verify(e):
    try:
        # 示例校验逻辑，用户自行替换为实际API请求逻辑
        resp = requests.post("https://your-verify-api.com", json={"data": e}, timeout=10)
        if resp.status_code == 200 and resp.json().get("valid"):
            # 加锁后写入set，保证线程安全
            with lock:
                valid_data.add(e)
    except Exception as e:
        logging.error(f"校验数据{e}失败: {str(e)}")

if __name__ == "__main__":
    try:
        with ThreadPoolExecutor(max_workers=5) as executor:
            run_threads = [executor.submit(data_verify, e) for e in s]
            # 等待所有线程执行完成
            for future in run_threads:
                future.result()
        # 正常执行完成保存数据
        save_data('ok_data.txt')
    except BaseException as e:
        # 捕获所有异常类型，包含Exception、KeyboardInterrupt、SystemExit等
        logging.error(f"程序异常终止: {str(e)}")
        save_data('error_save_data.txt')

注意事项 #

• requests请求建议添加timeout参数，避免单个请求卡死导致进程无法正常退出
• 若需支持Windows平台，可取消代码中SIGBREAK信号注册的注释
• 线程锁仅在写入valid_data时添加，不会对校验逻辑的执行效率产生明显影响

内容的提问来源于stack exchange，提问作者Calvin Danis