Celery幂等周期性任务：运行新实例前取消旧实例

要实现仅运行最新的周期性任务实例并取消旧实例，可以通过Redis存储活跃任务ID + 任务发布前撤销旧任务的方式实现，以下是具体方案：

实现思路 #

1. 用Redis存储当前处于排队或运行状态的任务ID，确保每次新任务触发时能定位到旧任务。
2. 每次调度新任务前，先撤销所有旧任务（包括排队中未启动的和正在运行的）。
3. 任务完成后清理Redis中的任务ID，避免残留无效数据。

完整代码实现 #

from celery import Celery
from celery.schedules import crontab
import time
import redis

# 初始化Celery应用
app = Celery('tasks', broker='redis://localhost:6379/0')

# 连接Redis，用于存储活跃任务ID
redis_client = redis.Redis(host='localhost', port=6379, db=0)
# 定义存储任务ID的Redis键名
ACTIVE_TASK_KEY = "my_periodic_task:active_id"

@app.task(bind=True)
def my_periodic_task(self):
    print(f"Starting task instance: {self.request.id}")
    try:
        # 模拟耗时任务
        time.sleep(30)
        print(f"Completed task instance: {self.request.id}")
    finally:
        # 任务完成后，仅当当前任务是活跃任务时清理Redis键
        current_active_id = redis_client.get(ACTIVE_TASK_KEY)
        if current_active_id and current_active_id.decode() == self.request.id:
            redis_client.delete(ACTIVE_TASK_KEY)

@app.on_after_configure.connect
def setup_periodic_tasks(sender, **kwargs):
    def schedule_latest_task():
        # 获取旧的活跃任务ID
        old_task_id = redis_client.get(ACTIVE_TASK_KEY)
        if old_task_id:
            old_task_id = old_task_id.decode()
            # 撤销旧任务：排队任务直接取消，运行中任务强制终止
            app.control.revoke(
                old_task_id,
                terminate=True,
                signal='SIGTERM'  # 用温和的终止信号，如需强制可改用'SIGKILL'
            )
            print(f"Revoked old task instance: {old_task_id}")
        
        # 发布新任务并记录ID到Redis
        new_task = my_periodic_task.apply_async()
        redis_client.set(ACTIVE_TASK_KEY, new_task.id)
        print(f"Scheduled new task instance: {new_task.id}")
    
    # 配置每10分钟执行一次调度逻辑
    sender.add_periodic_task(
        crontab(minute='*/10'),
        schedule_latest_task,
    )

关键细节说明 #

• 任务撤销逻辑：app.control.revoke会向所有Celery Worker广播撤销命令，确保多Worker环境下旧任务也能被处理。terminate=True会终止正在运行的任务，配合SIGTERM信号让任务尽量优雅退出。
• Redis存储作用：通过唯一的键名绑定当前活跃任务，避免每次调度时无法定位旧任务的问题。
• 幂等性保障：由于任务具备幂等性，即使旧任务被中途终止，新任务运行时也能完成预期操作，不会产生数据不一致问题。

注意事项 #

1. 确保Redis与Celery使用同一实例，避免连接或数据同步问题。
2. 如果任务需要更长时间运行，可在Redis的set操作中添加ex参数设置键的过期时间，防止意外情况导致键残留。
3. 若Worker节点较多，默认的广播模式已能覆盖大多数场景，如需指定目标Worker，可使用app.control.revoke的destination参数。

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