在Django REST Framework中用Celery实现指定任务顺序执行的方案咨询

核心思路 #

为每个请求生成唯一标识（request_id），用缓存记录对应first_function的执行状态，通过一个中间任务监听状态，待对应任务完成后再触发second_function——既保证first_function独立执行，又能在视图流程正常时实现顺序依赖，同时完全避免阻塞。

步骤1：生成请求唯一标识 #

在视图中为每个请求生成唯一request_id，用于关联两个任务：

# views.py
import uuid
from django.core.cache import cache
from django.http import Response
from .tasks import first_function, wait_for_first_then_run_second

def your_api_view(request):
    # 生成全局唯一的请求标识
    request_id = str(uuid.uuid4())
    
    # 立即触发first_function，不受后续代码异常影响
    first_function.delay(request_id)
    
    # --- 中间业务代码段 ---
    try:
        # 这里写你的核心业务逻辑：比如数据校验、数据库写入等
        # ...
        
        # 流程无异常时，触发中间任务，等待first完成后执行second
        wait_for_first_then_run_second.delay(request_id, "arg1", "arg2")  # 传入second_function需要的参数
    except Exception as e:
        # 异常处理逻辑：此时first_function已在后台运行，不受影响
        # ...
        pass
    
    return Response({"status": "accepted", "request_id": request_id})

步骤2：修改first_function记录执行状态 #

更新first_function，完成业务逻辑后在缓存中标记任务完成：

# tasks.py
from celery import shared_task
from django.core.cache import cache

@shared_task
def first_function(request_id):
    # 你的第一阶段业务逻辑
    # ...
    
    # 任务完成后，在缓存中写入标记（设置超时避免冗余）
    cache.set(f"task_done:{request_id}", True, timeout=3600)

步骤3：创建中间任务监听状态并触发second_function #

利用Celery的重试机制，创建中间任务轮询缓存状态，直到对应first_function完成后再执行second_function：

# tasks.py
@shared_task(bind=True, max_retries=30, retry_backoff=2)
def wait_for_first_then_run_second(self, request_id, *second_args, **second_kwargs):
    # 检查对应请求的first任务是否完成
    if cache.get(f"task_done:{request_id}"):
        # 执行second_function并传递参数
        second_function.delay(*second_args, **second_kwargs)
        # 删除缓存标记，避免重复触发
        cache.delete(f"task_done:{request_id}")
        return
    
    # 未完成则触发重试，间隔指数递增（2s→4s→8s...）
    self.retry(exc=Exception("First task not completed"), countdown=2)

@shared_task
def second_function(*args, **kwargs):
    # 你的第二阶段业务逻辑
    # ...

关键细节说明 #

• 请求隔离：每个请求的request_id唯一，缓存键与request_id绑定，确保中间任务只会等待对应请求的first_function完成，不会和其他请求的任务混淆。
• 非阻塞设计：中间任务的重试由Celery后台处理，视图无需阻塞等待，完全符合要求。
• 异常容错：
  • 若first_function执行失败，缓存不会写入完成标记，中间任务达到max_retries次数后自动停止，避免无限重试。
  • 视图后续代码抛出异常时，first_function已被触发，不受任何影响。
• 缓存选型：推荐配置Redis作为Django缓存后端，性能和可靠性远优于默认内存缓存，适配分布式部署场景。

替代方案：Celery事件监听（可选） #

若不想依赖缓存，可通过Celery事件系统监听first_function的完成事件，根据任务携带的request_id触发对应second_function。但该方案需要额外配置事件监听组件，复杂度较高，适合对缓存依赖敏感的分布式场景。

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