如何在Celery中基于参数实现指定任务按顺序执行

Celery 没有原生支持按自定义分组串行执行任务的开箱即用功能，但你无需从零开发完整的自定义队列管理方案，基于现有 Django + Celery 技术栈的内置特性即可快速实现需求。

这个方案最适配你现有10个worker的并发配置，同时能严格保证同一数据源的任务按提交顺序串行执行，不同数据源的任务不受影响保持并行处理。

实现逻辑 #

以数据源ID为唯一标识创建分布式锁，任务启动前先尝试获取对应数据源的锁：

• 拿到锁则正常执行任务，执行完成后释放锁
• 没拿到锁则延迟一段时间后重试，直到锁被释放后抢占执行

代码示例（基于Django缓存+Redis） #

你需要提前配置好Django的缓存后端为Redis（和Celery Broker共用Redis实例即可），安装django-redis依赖：

from celery import shared_task
from django.core.cache import cache
from django.conf import settings

@shared_task(bind=True, max_retries=None)
def process_csv(self, data_source_id: int, csv_path: str):
    # 锁超时时间建议设置为单个文件最大处理时长的2倍，避免任务异常退出导致死锁
    lock_key = f"csv_process:lock:{data_source_id}"
    lock = cache.lock(lock_key, timeout=settings.CSV_PROCESS_LOCK_TIMEOUT)
    lock_acquired = lock.acquire(blocking=False)

    if not lock_acquired:
        # 未拿到锁则延迟5秒重试，间隔可根据你的业务场景调整
        self.retry(countdown=5)
    
    try:
        # 此处替换为你原有CSV处理的业务逻辑
        your_original_csv_process_logic(csv_path)
    finally:
        if lock_acquired:
            lock.release()

• 动态专属队列：给每个数据源分配独立队列，提交任务时路由到对应队列，worker配置为从所有队列消费。该方案适合数据源数量较少的场景，你有600-700个数据源的场景下队列管理成本较高，不推荐。
• 任务链：同一数据源的新任务提交时，追加到上一个未完成的任务后面作为回调，通过Celery的chain原语实现串行。该方案需要额外维护每个数据源的最后一个任务ID，任务异常中断时容易出现后续任务无法执行的问题。

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