Python多线程运行schedule调度任务时如何处理错误不影响其他线程

• Python默认线程机制中，子线程抛出未捕获的普通异常时，只会终止当前子线程，不会直接影响其他运行中的线程。你遇到单任务崩溃连带全量任务停止的核心原因有两个：
  • 代码里的email()、writeFile()两个任务没有套run_threaded包装，直接运行在调度主循环所在的主线程中，这两个任务一旦抛出未捕获异常，会直接终止主循环，所有任务都会停止
  • 子线程内如果触发进程级致命错误（比如C扩展段错误、OOM内存溢出），会直接终止整个进程，连带所有线程退出，这类问题不管怎么捕异常都没用，需要从业务代码本身排查修复
• 原代码还有个笔误：fucntionE拼写错误，实际运行时会直接抛NameError

不需要在每个业务函数内部重复写异常捕获逻辑，在统一的线程入口层做一次全局异常拦截是最优方案，不会侵入业务代码，维护成本最低。

第一步：改造线程包装函数，统一做异常隔离 #

把异常捕获逻辑封装在run_threaded内部，所有通过这个方法启动的任务自动获得异常隔离能力，不需要修改业务函数：

import threading
import traceback
from time import sleep
import schedule

def run_threaded(job_func):
    def task_wrapper():
        try:
            job_func()
        except Exception as e:
            # 这里可以替换成你自己的日志、告警逻辑
            print(f"定时任务执行失败，错误信息：{str(e)}")
            traceback.print_exc()
    # 设置daemon=True，避免主线程退出时残留子线程阻塞进程
    job_thread = threading.Thread(target=task_wrapper, daemon=True)
    job_thread.start()

第二步：把主线程运行的任务也套入线程包装 #

原来直接在主线程执行的email()、writeFile()必须套入run_threaded，避免任务崩溃打挂主循环，同时修正拼写错误：

print('scheduler is starting')
schedule.every().day.at('06:30').do(run_threaded, lambda:functionA())
schedule.every().day.at('06:30').do(run_threaded, lambda:functionB())
schedule.every().day.at('06:30').do(run_threaded, lambda:functionC())
schedule.every().day.at('06:30').do(run_threaded, lambda:functionD())
schedule.every().day.at('06:30').do(run_threaded, lambda:functionE()) # 修正原代码fucntion拼写错误
schedule.every().day.at('22:00').do(run_threaded, lambda:email())
schedule.every().day.at('22:15').do(run_threaded, lambda:writeFile())

第三步：给主循环加兜底异常捕获 #

避免schedule框架本身的逻辑异常导致主循环退出：

while True:
    try:
        schedule.run_pending()
        sleep(1)
    except Exception as e:
        print(f"调度主循环异常：{str(e)}")
        traceback.print_exc()
        sleep(1) # 异常后等待1秒再重试，避免死循环占满CPU

如果你的定时任务可能出现执行时长超过调度间隔的情况，不建议裸开线程，换成线程池实现可以避免线程数无限膨胀耗尽系统资源：

from concurrent.futures import ThreadPoolExecutor
# 初始化全局线程池，最大线程数根据实际任务并发量调整
thread_pool = ThreadPoolExecutor(max_workers=8, thread_name_prefix="schedule_task")

def run_threaded(job_func):
    def task_wrapper():
        try:
            job_func()
        except Exception as e:
            print(f"定时任务执行失败，错误信息：{str(e)}")
            traceback.print_exc()
    thread_pool.submit(task_wrapper)

注意事项 #

• 异常捕获只需要拦截Exception层级即可，不要捕获BaseException，否则会把KeyboardInterrupt（Ctrl+C退出）、SystemExit这类系统级异常也吞掉，导致无法正常终止程序
• 异常处理块不要留空（不要直接写pass），至少要打印完整错误栈，否则任务静默失败很难排查问题
• 如果任务涉及文件、数据库连接等资源操作，建议在异常块里增加资源释放逻辑，避免资源泄漏

内容的提问来源于stack exchange，提问作者Evans Gunawan