如何将Celery定时任务的traceback输出至控制台或日志文件

普通Celery任务的异常捕获方案通常只覆盖了worker执行层的异步任务，定时任务由Celery Beat调度派发，异常链路经过Beat调度层、worker执行层两层，很多通用方案没有覆盖调度层配置，所以会出现不生效的情况，以下是经过验证的可行方案：

方案1：全局配置Celery日志及异常传播（无侵入，无需改任务代码） #

• 调整Celery启动参数，强制开启traceback打印：
  如果是Beat和Worker合并启动，启动命令加上--loglevel=INFO --traceback参数，示例：

  celery -A 你的项目模块名 worker -B --loglevel=INFO --traceback
  

  如果是Beat和Worker分开部署，两边都要配置日志级别，Beat单独启动命令示例：

  celery -A 你的项目模块名 beat --loglevel=INFO
  

• 在Celery配置文件中添加以下配置，完整输出堆栈信息：

  # 自定义日志格式，强制携带异常堆栈信息
  CELERY_WORKER_LOG_FORMAT = '[%(asctime)s: %(levelname)s/%(processName)s] %(task_id)s %(task_name)s %(message)s %(exc_info)s'
  CELERY_WORKER_TASK_LOG_FORMAT = '[%(asctime)s: %(levelname)s/%(processName)s] %(task_id)s %(task_name)s %(message)s %(exc_info)s'
  # 禁用异常信息截断
  CELERY_MAX_CACHED_RESULTS = -1
  CELERY_TASK_TRACK_STARTED = True
  

方案2：定时任务统一装饰器捕获（适合自定义日志输出场景） #

如果需要把traceback输出到指定日志文件，或者叠加异常告警逻辑，可以给所有定时任务加统一异常捕获装饰器：

from functools import wraps
import traceback
import logging

# 可自行配置logger输出到指定文件、控制台等
logger = logging.getLogger('celery_schedule_task')

def schedule_task_exception_catch(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            # 输出完整traceback
            logger.error(f"定时任务[{func.__name__}]执行异常：{str(e)}\n堆栈信息：\n{traceback.format_exc()}")
            # 重新抛出异常，不影响Celery原生重试、状态标记逻辑
            raise
    return wrapper

装饰器需要放在@app.task装饰器下方，示例：

@app.task
@schedule_task_exception_catch
def your_schedule_task():
    # 任务业务逻辑
    pass

方案3：监听Celery任务失败信号（全局无侵入方案） #

不需要修改任何任务代码，直接监听Celery原生的任务失败信号，所有任务（含定时任务）异常时都会自动打印traceback：

from celery import signals
import traceback
import logging

logger = logging.getLogger(__name__)

@signals.task_failure.connect
def on_task_failure(sender, task_id, exception, traceback_obj, **kwargs):
    logger.error(f"""
    任务执行失败：
    任务名称：{sender.name}
    任务ID：{task_id}
    异常信息：{str(exception)}
    完整堆栈：\n{''.join(traceback.format_tb(traceback_obj))}
    """)

把这段代码放到Celery实例初始化的文件中即可生效。

排查注意：如果以上方案仍不生效，检查定时任务的@app.task装饰器是否加了ignore_result=True配置，该配置会丢弃异常信息，改为False后重新测试即可。

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