Python 分布式任务队列重复执行解决方案

在分布式任务队列中确保任务不重复执行，需要结合幂等性设计、唯一标识、状态追踪和分布式锁等多种机制。以下是用Python实现的具体方案：

一、核心解决思路

1. 任务幂等性设计：确保任务重复执行不会影响最终结果。
2. 唯一任务标识 (Unique Task ID)：为每个任务生成唯一ID，处理前校验是否已执行。
3. 原子性操作：利用数据库唯一约束或Redis的SETNX命令实现原子性校验。
4. 消息队列确认机制：正确处理任务后才发送ACK，避免消息重新入队。
5. 分布式锁：控制同一任务同时只能被一个Worker处理。

二、Python实现步骤

1. 生成唯一任务ID

import uuid

def create_task(task_data):
    task_id = str(uuid.uuid4())  # 生成唯一ID
    # 将task_id和task_data存储到消息队列和数据库
    return task_id

2. 任务处理前校验（Redis原子操作）

import redis

redis_client = redis.StrictRedis()

def process_task(task_id, task_data):
    # 原子性检查任务是否已处理
    if redis_client.setnx(f"task:{task_id}", "processing"):
        try:
            # 执行任务逻辑...
            # 标记任务为已完成
            redis_client.setex(f"task:{task_id}", 3600, "completed")
            return True
        except Exception as e:
            # 处理失败，删除锁允许重试
            redis_client.delete(f"task:{task_id}")
            raise e
    else:
        # 任务已存在，直接跳过
        return False

3. 数据库唯一约束（防止重复插入）

# 使用SQLAlchemy示例
from sqlalchemy import Column, String, UniqueConstraint

class Task(Base):
    __tablename__ = 'tasks'
    id = Column(String(36), primary_key=True)
    status = Column(String(20))
    __table_args__ = (UniqueConstraint('id'),)

def save_task(session, task_id):
    try:
        task = Task(id=task_id, status='pending')
        session.add(task)
        session.commit()
    except IntegrityError:
        session.rollback()
        raise DuplicateTaskError("任务已存在")

4. 消息队列确认机制（以RabbitMQ为例）

import pika

def start_worker():
    connection = pika.BlockingConnection(pika.ConnectionParameters('localhost'))
    channel = connection.channel()
    channel.queue_declare(queue='task_queue', durable=True)
    
    def callback(ch, method, properties, body):
        task_id = extract_task_id(body)
        if not is_task_processed(task_id):  # 检查任务是否已处理
            process_task(task_id, body)
            ch.basic_ack(delivery_tag=method.delivery_tag)  # 确认消息
        else:
            ch.basic_reject(delivery_tag=method.delivery_tag, requeue=False)  # 丢弃消息
    
    channel.basic_consume(queue='task_queue', on_message_callback=callback)
    channel.start_consuming()

5. 分布式锁（使用RedLock算法）

from redis import Redis
from redis.lock import RedLock

def process_task_with_lock(task_id):
    lock = RedLock(f"lock:{task_id}", [Redis()], ttl=60000)
    if lock.acquire():
        try:
            if not is_task_processed(task_id):
                execute_task(task_id)
        finally:
            lock.release()

三、方案对比与选择

四、增强健壮性措施

1. 设置任务状态机：记录任务状态（pending/running/completed），处理前校验状态。
2. 超时和重试机制：为任务设置合理超时时间，避免无限阻塞。
3. 异步清理机制：定期清理已完成的任务记录，避免存储膨胀。

# 任务状态机示例
TASK_STATES = ['pending', 'running', 'completed', 'failed']

def update_task_state(session, task_id, new_state):
    task = session.query(Task).get(task_id)
    if task.state == 'completed':
        raise InvalidStateTransitionError("任务已完成，不可修改")
    task.state = new_state
    session.commit()

五、完整架构示意图

[Client] --> (生成唯一Task ID) --> [消息队列]
                             ↓
                    [Worker集群] --> [检查Redis/Database] --> [执行任务]
                             ↓
                    [确认消息] --> [标记任务状态]

通过组合上述方法，可有效避免分布式环境中的任务重复执行问题。实际选择时需根据业务需求权衡一致性和性能。