Flask集成Dramatiq如何异步实现任务串行、避免未完成重复触发

核心逻辑是用分布式锁按业务标识（这里对应任务参数里的loan_number，相同贷款号的上传任务属于需要串行的同类型任务）标记任务运行状态，拿到锁的任务正常执行，未拿到锁的任务自动延迟重试，全程无IO阻塞，不会影响Flask服务响应。

实现步骤 #

1. 准备分布式锁存储 #

推荐用Redis实现分布式锁，首先安装Redis依赖：

pip install redis

在项目公共配置中初始化Redis客户端：

import redis
# 替换为你实际的Redis连接参数
redis_client = redis.Redis(host='127.0.0.1', port=6379, db=0, decode_responses=True)

2. 修改Actor任务添加锁逻辑 #

import dramatiq
import os

# 按需配置锁过期时间，必须大于单任务最长预计执行时间，示例为1小时
UPLOAD_TASK_LOCK_EXPIRE = 3600

@dramatiq.actor(store_results=True, max_retries=20)
def my_actor(credentials, loan_number, id_folder):
    # 按loan_number生成唯一锁key，隔离不同业务的任务
    task_lock_key = f"gdrive_upload_lock:{loan_number}"
    # 原子性尝试获取锁：nx表示仅key不存在时才设置成功，ex设置自动过期时间
    get_lock = redis_client.set(task_lock_key, "running", ex=UPLOAD_TASK_LOCK_EXPIRE, nx=True)

    # 未获取到锁说明同类型任务正在运行，生成延迟重试任务后直接返回
    if not get_lock:
        my_actor.send_with_options(
            args=(credentials, loan_number, id_folder),
            delay=30 * 1000  # 重试延迟30秒，单位为毫秒，可按需调整
        )
        return

    try:
        print('start')
        g = GoogleApi(credentials)
        directory = os.path.join(UPLOAD_FOLDER_MAIN, str(loan_number))
        # 注意：原代码中的list_of_files建议改为函数内生成或作为参数传入，避免全局变量并发冲突
        for i in list_of_files:
            g.upload_file(i, id_folder)
        return True
    finally:
        # 任务执行完成（无论成功失败）都主动释放锁
        redis_client.delete(task_lock_key)

3. 调用方式 #

Flask端直接异步发送任务即可，不需要加任何阻塞等待逻辑：

my_actor.send(credentials, loan_number, id_folder)

方案说明 #

• 全程无阻塞：Flask发送任务后立即返回，Worker端未拿到锁时直接生成延迟任务就释放资源，不会占用进程线程资源，完全符合异步非阻塞要求。
• 分布式场景兼容：即使部署多个Dramatiq Worker节点，分布式锁也能保证同类型任务串行执行。
• 异常兼容：锁自带过期时间，就算任务异常退出没有主动释放锁，到期后也会自动释放，不会出现死锁。

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