Celery任务未执行排查：Worker与Beat运行正常但无输出

问题背景 #

原Flask应用从APScheduler切换到Celery，Redis配置正常，Beat和Worker启动后任务已被接收（日志显示），但任务内的print语句无任何输出，已尝试移除debug模式、重启服务、测试简单任务等操作，问题仍存在。

排查与解决方案 #

1. 调整Celery Worker日志级别与输出捕获 #

默认情况下，print的标准输出不会被Celery的info级别日志捕获，需调整日志级别或开启stdout日志：

• 启动Worker时使用debug级别日志：

  celery -A app.celery worker --loglevel=debug
  

• 或在Celery配置中强制捕获标准输出：

  celery.conf.update({
      'worker_stdout_logging': True,
      'worker_stderr_logging': True
  })
  

2. 确保任务内激活Flask应用上下文 #

check_expiry任务未激活Flask上下文直接操作数据库，可能导致静默异常（即使有try-except，也可能因上下文缺失无法正常执行或输出），需添加上下文：

@celery.task
def check_expiry():
    try:
        with app.app_context():  # 添加Flask应用上下文
            print("Executing check_expiry task at ", datetime.now())
            issued = requests_db.query.filter_by(status='issued').all()
            for issue in issued:
                if issue.end_date and issue.end_date < datetime.now():
                    print(f"Issue {issue.id} is expired.")
                    issue.status = "completed"
            db.session.commit()
            print("check_expiry task completed.")
    except Exception as e:
        print(f"Error in check_expiry task: {e}")

3. 替换print为Celery专用日志系统 #

print在Celery任务中输出不稳定，建议使用Celery内置日志，确保输出能被Worker日志捕获：

from celery.utils.log import get_task_logger
from datetime import datetime

logger = get_task_logger(__name__)

@celery.task
def test_celery():
    logger.info("Celery task executed at %s", datetime.now())

@celery.task
def check_expiry():
    try:
        with app.app_context():
            logger.info("Executing check_expiry task at %s", datetime.now())
            issued = requests_db.query.filter_by(status='issued').all()
            for issue in issued:
                if issue.end_date and issue.end_date < datetime.now():
                    logger.info("Issue %d is expired.", issue.id)
                    issue.status = "completed"
            db.session.commit()
            logger.info("check_expiry task completed.")
    except Exception as e:
        logger.error("Error in check_expiry task: %s", str(e))

4. 验证虚拟环境一致性 #

确保启动Worker、Beat和Flask应用的是同一个虚拟环境，避免依赖版本不一致导致的静默执行失败：

• 激活虚拟环境后再启动服务：

  source venv/bin/activate  # Linux/macOS
  venv\Scripts\activate     # Windows
  celery -A app.celery worker --loglevel=debug
  celery -A app.celery beat --loglevel=info
  

5. 手动触发任务验证 #

在Flask shell中手动触发任务，验证任务是否真的执行：

# 进入Flask shell
flask shell
# 手动调用任务
from app import test_celery
test_celery.delay()

观察Worker终端是否有输出，若仍无输出，需检查Redis连接是否正常，或Celery与Flask的配置是否冲突。

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