如何避免不同Airflow DAG中相同任务的并发执行？

核心需求 #

特定任务在所有DAG中同时仅允许1个实例运行，无需明确依赖流向，保留原有调度逻辑。

替代方案推荐 #

1. 基于Airflow原生分布式锁实现（推荐） #

利用Airflow内置的锁机制替代自定义变量锁，从根源避免并发更新时的竞态问题：

• 实现思路：任务启动前，以目标任务ID为锁名尝试获取全局锁；获取成功后检查是否有同任务的其他运行实例，确认唯一后执行任务，执行完毕自动释放锁。
• 代码示例：

from airflow.utils.db import provide_session
from airflow.models import TaskInstance
from airflow.utils.lock import Lock
from airflow.exceptions import AirflowException

def acquire_global_task_lock(task_id):
    lock_id = f"global_task_mutex_{task_id}"
    with Lock(lock_id, timeout=60):
        # 查询所有DAG中该任务的运行实例
        running_instances = TaskInstance.find_running_task_instances(task_id=task_id)
        if len(running_instances) > 1:
            return False
        return True

def target_task_logic(**context):
    task_id = context["task"].task_id
    
    if not acquire_global_task_lock(task_id):
        raise AirflowException("同任务在其他DAG中运行，等待重试")
    
    # 任务核心执行逻辑
    print("执行任务核心操作...")

• 优势：依赖Airflow原生组件，无需额外服务，锁机制稳定可靠。

2. 借助外部分布式锁服务（Redis/ZooKeeper） #

针对大规模Airflow集群，可使用成熟的外部锁服务实现更灵活的互斥控制：

• 实现思路：任务执行前连接锁服务，尝试获取带超时的原子锁；获取成功则执行任务，失败则触发重试；任务结束后主动释放锁。
• 代码示例（Redis）：

import redis
from airflow.models import Variable
from airflow.exceptions import AirflowException

def locked_task_logic(**context):
    task_id = context["task"].task_id
    redis_host = Variable.get("redis_host")
    lock_key = f"task_mutex_{task_id}"
    
    redis_client = redis.Redis(host=redis_host, port=6379)
    # 设置锁，超时3600秒防止死锁
    lock_acquired = redis_client.set(lock_key, "active", nx=True, ex=3600)
    
    if not lock_acquired:
        raise AirflowException("同任务已在其他DAG运行，等待重试")
    
    try:
        # 任务核心逻辑
        print("执行任务核心操作...")
    finally:
        redis_client.delete(lock_key)

• 优势：锁机制成熟，支持自动超时释放，适合复杂集群场景。

3. 封装互斥逻辑为自定义Operator #

将互斥控制逻辑封装成可复用的Operator，简化多DAG中的代码重复：

from airflow.models.baseoperator import BaseOperator
from airflow.utils.lock import Lock
from airflow.models import TaskInstance
from airflow.exceptions import AirflowException

class MutexTaskOperator(BaseOperator):
    def __init__(self, target_task_id, **kwargs):
        super().__init__(**kwargs)
        self.target_task_id = target_task_id

    def execute(self, context):
        lock_id = f"global_mutex_{self.target_task_id}"
        with Lock(lock_id, timeout=60):
            running_instances = TaskInstance.find_running_task_instances(task_id=self.target_task_id)
            if len(running_instances) > 1:
                self.log.error("同任务在其他DAG运行，触发重试")
                raise AirflowException("互斥锁未获取")
            
            # 执行任务核心逻辑
            self.log.info("执行任务核心操作")

• 使用方式：在各DAG中直接实例化该Operator，传入目标任务ID即可。

与自定义Airflow变量方案的对比 #

你提出的变量锁方案存在竞态风险：当多个任务同时读取变量为"未锁定"，并同时更新为"锁定"时，会导致多实例同时执行。而上述方案通过分布式锁或原子操作避免了该问题，可靠性更高。

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