Python定时任务中的异常处理与日志记录

引言 / 什么是Python定时任务中的异常处理与日志记录

在Python自动化开发中，定时任务（如使用schedule、APScheduler或Celery等库）是常见的需求。无论是数据同步、报表生成还是系统监控，定时任务都需要长期稳定运行。然而，网络波动、资源不足或代码逻辑错误等意外情况可能导致任务失败，若缺乏异常处理机制，程序可能直接崩溃且无法追溯问题根源。

异常处理能捕获并处理运行时错误，避免任务中断；日志记录则能完整记录任务执行过程，为问题排查提供依据。两者结合可显著提升定时任务的健壮性。本文将以schedule库为例，详细讲解如何实现优雅的异常处理与日志记录。

准备工作

1. 环境要求：Python 3.6+（推荐3.8+），需安装schedule库（pip install schedule）。
2. 基础代码结构：以下是一个简单的定时任务框架：

   import schedule
   import time
   
   def job():
       print("任务执行中...")
       # 模拟任务逻辑
       time.sleep(1)
   
   schedule.every(10).seconds.do(job)
   
   while True:
       schedule.run_pending()
       time.sleep(1)
   

基础操作：异常处理与日志记录

步骤一：配置日志记录

Python内置的logging模块支持多级别日志（DEBUG/INFO/WARNING/ERROR/CRITICAL）和输出到文件/控制台。以下是一个基础配置：

import logging
from logging.handlers import RotatingFileHandler

def setup_logger():
    logger = logging.getLogger("timed_task")
    logger.setLevel(logging.INFO)  # 设置日志级别

    # 控制台输出
    console_handler = logging.StreamHandler()
    console_handler.setFormatter(
        logging.Formatter("%(asctime)s - %(name)s - %(levelname)s - %(message)s")
    )
    logger.addHandler(console_handler)

    # 文件输出（按大小滚动）
    file_handler = RotatingFileHandler(
        "timed_task.log", maxBytes=1024*1024, backupCount=3
    )
    file_handler.setFormatter(
        logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
    )
    logger.addHandler(file_handler)

    return logger

logger = setup_logger()

步骤二：在定时任务中捕获异常

使用try-except块包裹任务逻辑，捕获异常后记录日志：

def job_with_exception_handling():
    try:
        logger.info("任务开始执行")
        # 模拟可能出错的操作
        result = 10 / 0  # 触发ZeroDivisionError
        logger.info(f"任务结果: {result}")
    except ZeroDivisionError as e:
        logger.error(f"数学运算错误: {str(e)}", exc_info=True)  # exc_info记录堆栈
    except Exception as e:
        logger.critical(f"未知错误: {str(e)}", exc_info=True)
    finally:
        logger.info("任务执行结束")

schedule.every(10).seconds.do(job_with_exception_handling)

提示：exc_info=True会记录完整的错误堆栈，便于定位问题。生产环境中建议对不同异常类型分别处理。

进阶技巧

技巧一：全局异常装饰器

若多个任务共享相同异常处理逻辑，可使用装饰器减少重复代码：

def handle_exceptions(func):
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            logger.error(f"任务 {func.__name__} 执行失败: {str(e)}", exc_info=True)
    return wrapper

@handle_exceptions
def complex_job():
    logger.info("复杂任务开始")
    # 业务逻辑...
    return "成功"

schedule.every(20).seconds.do(complex_job)

技巧二：日志分级与过滤

根据任务重要性设置不同日志级别，例如：

def critical_job():
    logger.critical("高优先级任务触发")  # 强制记录
    # ...

def debug_job():
    if logger.isEnabledFor(logging.DEBUG):  # 仅在DEBUG级别记录
        logger.debug("调试信息: 参数校验通过")

技巧三：集成邮件/短信告警

当捕获严重错误时，可通过smtplib或第三方服务（如Sentry）发送告警：

import smtplib
from email.message import EmailMessage

def send_alert(error_msg):
    msg = EmailMessage()
    msg["Subject"] = "定时任务错误告警"
    msg["From"] = "admin@example.com"
    msg["To"] = "ops@example.com"
    msg.set_content(f"错误详情:\n{error_msg}")

    with smtplib.SMTP("smtp.example.com", 587) as server:
        server.starttls()
        server.login("user", "password")
        server.send_message(msg)

# 在异常处理中调用
except Exception as e:
    error_msg = f"任务失败: {str(e)}"
    logger.critical(error_msg, exc_info=True)
    send_alert(error_msg)

技巧四：日志分析案例

假设日志文件内容如下：

2026-05-04 10:00:00,123 - INFO - 任务开始执行
2026-05-04 10:00:00,456 - ERROR - 数学运算错误: division by zero
Traceback (most recent call last):
  File "task.py", line 15, in job_with_exception_handling
    result = 10 / 0
ZeroDivisionError: division by zero
2026-05-04 10:00:00,789 - INFO - 任务执行结束

通过分析日志，可快速定位到ZeroDivisionError发生在job_with_exception_handling函数的除法操作中，进而修改代码或添加数据校验。

常见问题

Q：如何避免日志文件过大？

A：使用RotatingFileHandler（按大小滚动）或TimedRotatingFileHandler（按时间滚动）。示例：

# 每天生成一个新日志文件，保留7天
handler = TimedRotatingFileHandler(
    "timed_task.log", when="midnight", interval=1, backupCount=7
)

Q：日志记录会影响定时任务性能吗？

A：日志写入通常采用异步方式（如QueueHandler），对主任务影响较小。若对性能敏感，可：

1. 仅在ERROR及以上级别记录堆栈
2. 将日志输出到内存缓冲区（StringIO）后批量写入

Q：如何同时输出日志到控制台和文件？

A：在setup_logger()函数中同时添加StreamHandler和FileHandler即可（如本文基础配置所示）。

小结

本文通过代码示例和场景分析，详细讲解了Python定时任务中异常处理与日志记录的实现方法：

1. 使用logging模块配置多级别、多输出的日志系统
2. 通过try-except捕获异常并记录详细错误信息
3. 借助装饰器、邮件告警等技巧提升可维护性
4. 结合日志分析快速定位问题

实际开发中，建议将异常处理与日志记录封装为独立模块，便于在多个项目中复用。完整代码示例可参考以下结构：

# task_manager.py
import schedule
import time
import logging
from logging.handlers import RotatingFileHandler

class TaskManager:
    def __init__(self):
        self.logger = self._setup_logger()

    def _setup_logger(self):
        logger = logging.getLogger("TaskManager")
        logger.setLevel(logging.INFO)
        # ...（同前文配置）
        return logger

    def add_job(self, interval, job_func):
        wrapped_job = self._wrap_job(job_func)
        schedule.every(interval).seconds.do(wrapped_job)

    def _wrap_job(self, func):
        def wrapper():
            try:
                self.logger.info(f"执行任务: {func.__name__}")
                result = func()
                self.logger.info(f"任务完成: {func.__name__}, 结果: {result}")
            except Exception as e:
                self.logger.error(
                    f"任务失败: {func.__name__}, 错误: {str(e)}",
                    exc_info=True
                )
        return wrapper

    def run(self):
        while True:
            schedule.run_pending()
            time.sleep(1)

# 使用示例
if __name__ == "__main__":
    manager = TaskManager()

    def sample_job():
        return 10 / 2  # 可修改为 10/0 测试异常

    manager.add_job(10, sample_job)
    manager.run()