Celery任务部分代码未执行无报错及结果缓存方案咨询

• 最常见的原因是修改任务代码后未重启Celery Worker进程：Celery Worker启动时会一次性加载所有任务代码，后续你对task函数的修改（包括新增的try/catch逻辑、run函数调用逻辑）都不会在运行中的Worker上生效，Worker运行的还是旧版本的代码，自然不会输出你新增的日志。
• 日志输出路径不匹配：你查看的是Django进程的日志文件，但Celery Worker是独立运行的进程，如果你没有单独配置Worker的日志输出到同一个文件，Worker内的logger输出默认会打印到Worker进程的控制台，不会写入Django的日志文件。你可以启动Worker时加上--logfile参数指定日志路径，比如celery -A 你的项目名 worker --loglevel=CRITICAL --logfile=./celery.log，再查看对应日志文件确认输出。
• 任务触发了超时被强制终止：如果你配置了Celery任务的time_limit/soft_time_limit，且run函数的执行时间超过了限制，Worker会直接终止当前任务进程，不会触发你写的通用异常捕获，也不会输出后续日志。你可以先取消任务的超时配置，再测试确认是否是该问题。

• 第一步：配置Celery的结果后端为Redis，在Django settings中添加配置：

  CELERY_RESULT_BACKEND = 'redis://127.0.0.1:6379/0'
  # 可选配置结果过期时间，避免Redis内存占用过高
  CELERY_TASK_RESULT_EXPIRES = 3600 # 结果保留1小时
  

• 第二步：视图中调用异步任务时，保存返回的任务ID：

  def view(request):
      logger.critical("before running celery")
      task_result = task.delay()
      # 把任务ID存入session，或者返回给前端存在隐藏字段中
      request.session['task_id'] = task_result.id
      return HttpResponse('任务已提交')
  

• 第三步：新增一个查询任务结果的视图，供前端轮询调用：

  from celery.result import AsyncResult
  def check_task_status(request):
      task_id = request.session.get('task_id')
      if not task_id:
          return JsonResponse({'status': 'not_found'})
      result = AsyncResult(task_id)
      if result.ready():
          # 任务执行完成，返回结果
          return JsonResponse({'status': 'success', 'data': result.result})
      else:
          return JsonResponse({'status': 'running'})
  

• 第四步：前端页面通过定时轮询check_task_status接口，拿到结果后渲染到HTML中即可。如果不想用轮询，也可以结合Django Channels实现WebSocket主动推送结果。

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