如何实现Celery任务顺序执行 避免多解析器运行引发的媒体文件问题

以下是可直接落地的4种实现方案，可根据你的业务场景选择：

方案1：专属队列+限制worker并发（最稳定，无需修改任务逻辑） #

该方案从调度底层强制同一时间只能有一个解析任务执行：

• 给XML解析任务绑定专属队列，任务定义时指定队列参数：

@shared_task(queue="xml_parse_queue")
def parse_xml(xml_file_path):
    # 原有解析存库逻辑
    pass

• 启动绑定该队列的worker时，将并发数设为1：
  celery -A your_project_name worker --queue=xml_parse_queue --concurrency=1 -l info
• 优势：完全原生支持，不存在并行风险，适合对稳定性要求高的场景。

方案2：任务链（Chain）按顺序调度 #

如果待执行的解析任务可以提前归集，直接用Celery原生的任务链实现串行，上一个任务执行完成后自动触发下一个：

from celery import chain

# 按顺序传入待执行的任务，参数为对应XML文件路径
parse_task_chain = chain(
    parse_xml.s("/data/file1.xml"),
    parse_xml.s("/data/file2.xml"),
    parse_xml.s("/data/file3.xml")
)
# 异步触发任务链
parse_task_chain.apply_async()

如果需要每个任务执行完成后加固定延迟再执行下一个，可在任务逻辑末尾通过apply_async(countdown=N)手动触发后续任务。

方案3：分布式锁兜底，避免同类型任务并行 #

适合解析任务来自多个提交入口、无法提前做任务串联的场景，用Redis做分布式锁保证同一时间只有一个解析任务能运行：

import redis
from celery import shared_task

redis_client = redis.Redis(host="127.0.0.1", port=6379, db=0)
PARSE_TASK_LOCK_KEY = "xml_parse_running_lock"
# 锁过期时间设为单任务最大执行时长的2倍，避免任务异常挂掉导致死锁
LOCK_EXPIRE_SECOND = 1800

@shared_task(bind=True, max_retries=10)
def parse_xml(self, xml_file_path):
    # 尝试抢占锁
    get_lock = redis_client.set(PARSE_TASK_LOCK_KEY, "1", ex=LOCK_EXPIRE_SECOND, nx=True)
    if not get_lock:
        # 没抢到锁则10秒后重试，可根据需求调整延迟时长
        self.retry(countdown=10)
    try:
        # 原有解析存库逻辑
        parse_and_save_to_db(xml_file_path)
    finally:
        # 任务执行完成后主动释放锁
        if redis_client.get(PARSE_TASK_LOCK_KEY) == b"1":
            redis_client.delete(PARSE_TASK_LOCK_KEY)

方案4：任务提交时设置延迟间隔 #

如果你只是需要避免任务集中执行给系统带来压力，提交任务时直接给每个任务设置错开的执行时间即可：

# 假设要提交3个任务，每个间隔30秒执行
parse_xml.apply_async(args=["/data/file1.xml"], countdown=0)
parse_xml.apply_async(args=["/data/file2.xml"], countdown=30)
parse_xml.apply_async(args=["/data/file3.xml"], countdown=60)

注意事项 #

• 所有方案都建议给解析任务添加幂等性校验，避免重试时重复写入数据或者重复操作媒体文件
• 单worker队列方案需要配套任务积压监控，避免任务量过大时队列阻塞未及时发现
• 分布式锁方案不要省略过期时间配置，生产环境建议同时配置worker任务超时强制杀掉的规则，避免锁长时间不释放

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