Django+Redis+Celery+Celery Beat配置异常：测试任务无日志输出求助

我之前也踩过类似的坑，明明严格跟着官方文档操作，但任务就是没动静，这种情况确实让人头疼。咱们一步步来拆解排查：

1. 先确认基础连接是否正常 #

• 验证Redis可用性：在终端执行 redis-cli ping，如果返回 PONG 说明Redis服务正常；如果没响应，先重启Redis服务（比如 systemctl restart redis 或 brew services restart redis）。
• 检查Django配置：确保settings.py里的Celery相关配置正确指向Redis：

  CELERY_BROKER_URL = 'redis://localhost:6379/0'
  CELERY_RESULT_BACKEND = 'redis://localhost:6379/0'
  # 针对Celery Beat的调度器配置
  CELERY_BEAT_SCHEDULER = 'django_celery_beat.schedulers:DatabaseScheduler'
  

• 别忘了执行 python manage.py migrate django_celery_beat，生成Celery Beat需要的数据库表。

2. 检查服务启动命令的正确性 #

• Celery Worker启动：必须在项目根目录下执行，命令应为：

  celery -A aenima worker -l info
  

  启动后看日志里有没有 Connected to redis://localhost:6379/0 这类连接成功的提示；如果有Task not registered的报错，说明Celery没找到你的任务。
• Celery Beat启动：你的命令后半段被截断了，正确的完整命令应该是：

  celery -A aenima beat -l debug --scheduler django_celery_beat.schedulers:DatabaseScheduler
  

  启动后看debug日志里有没有Scheduler: Sending task xxx的条目，这说明Beat在往队列里发任务。

3. 确认任务能被Celery发现 #

• 确保tasks.py放在你已经注册到INSTALLED_APPS的Django app目录下，比如myapp/tasks.py。
• 在项目根目录的celery.py里加上自动发现任务的配置：

  from celery import Celery
  from django.conf import settings
  
  app = Celery('aenima')
  app.config_from_object('django.conf:settings', namespace='CELERY')
  # 自动发现所有已注册app里的tasks
  app.autodiscover_tasks(lambda: settings.INSTALLED_APPS)
  

• 任务定义要规范，比如：

  from celery import shared_task
  import logging
  
  logger = logging.getLogger(__name__)
  
  @shared_task
  def test_periodic_task():
      logger.info("定时任务已执行！")
      print("定时任务已执行！")
      return "Task Done"
  

4. 检查日志配置是否生效 #

• 确保Django的settings.py里配置了Celery的日志输出：

  LOGGING = {
      'version': 1,
      'disable_existing_loggers': False,
      'handlers': {
          'console': {'class': 'logging.StreamHandler'},
      },
      'loggers': {
          'celery': {
              'handlers': ['console'],
              'level': 'INFO',
              'propagate': True,
          },
      },
  }
  

  这样Celery的日志会直接输出到终端，方便你排查。

5. 先手动触发任务，缩小问题范围 #

先跳过Celery Beat，直接在Django Shell里手动触发任务：

python manage.py shell

然后执行：

from myapp.tasks import test_periodic_task
result = test_periodic_task.delay()
print(result.status)  # 正常会从PENDING变为SUCCESS
print(result.get())  # 应该返回"Task Done"

• 如果手动触发有日志/打印输出，说明问题出在Celery Beat的周期性任务配置上（比如admin里的Periodic Task没正确关联任务、执行周期设置错误）；
• 如果手动触发也没输出，那问题大概率在Celery Worker和Redis的连接，或者任务定义本身。

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