如何用SQL表行项实现多服务器任务队列并避免重复执行

一、修改PostgreSQL代码避免任务重复执行 #

原代码的核心问题是查询与更新操作分离，存在竞态条件：多台服务器可能同时查到同一个state=0的任务，随后都执行更新，导致重复处理。要解决这个问题，必须把"获取任务"和"标记为处理中"做成原子操作，利用PostgreSQL的事务和行级锁特性实现。

方案1：原子更新+返回结果（推荐） #

通过UPDATE ... RETURNING语句直接在原子操作中完成任务锁定与获取，确保只有一台服务器能拿到目标任务：

import sqlalchemy
from my_engines import STATE_ENGINE
from my_functions import my_function, save_return_value_sql_db

# 开启事务，保证所有操作的原子性
with STATE_ENGINE.begin() as conn:
    # 原子更新并返回被选中的任务，避免多实例竞争
    result = conn.execute(
        sqlalchemy.text("""
            UPDATE states 
            SET state = 1 
            WHERE args = (
                SELECT args FROM states WHERE state = 0 ORDER BY args ASC LIMIT 1
            )
            RETURNING args
        """)
    ).fetchone()
    
    if result:
        current_arg = result[0]
        # 执行任务
        return_value = my_function(current_arg)
        save_return_value_sql_db(return_value)
        # 可选：任务完成后标记为已完成（state=2）
        conn.execute(
            sqlalchemy.text("UPDATE states SET state = 2 WHERE args = :arg"),
            {"arg": current_arg}
        )

方案2：行级锁+跳过已锁行 #

利用FOR UPDATE SKIP LOCKED锁定未被处理的任务，跳过已被其他实例锁定的行，同样能避免重复：

import sqlalchemy
from my_engines import STATE_ENGINE
from my_functions import my_function, save_return_value_sql_db

with STATE_ENGINE.begin() as conn:
    # 锁定未处理任务，跳过已被其他实例锁定的行
    result = conn.execute(
        sqlalchemy.text("""
            SELECT args FROM states 
            WHERE state = 0 
            ORDER BY args ASC 
            LIMIT 1 
            FOR UPDATE SKIP LOCKED
        """)
    ).fetchone()
    
    if result:
        current_arg = result[0]
        # 标记为处理中
        conn.execute(
            sqlalchemy.text("UPDATE states SET state = 1 WHERE args = :arg"),
            {"arg": current_arg}
        )
        # 执行任务
        return_value = my_function(current_arg)
        save_return_value_sql_db(return_value)
        # 标记为已完成
        conn.execute(
            sqlalchemy.text("UPDATE states SET state = 2 WHERE args = :arg"),
            {"arg": current_arg}
        )

关键注意事项 #

• 必须在事务中执行上述操作，否则行级锁会立即释放，无法起到排他作用；
• 建议给states表的state和args字段建立联合索引，提升查询与更新的效率；
• 可增加last_updated字段，定期清理超时未完成的任务（比如state=1但超过1小时未更新为state=2的任务），避免任务卡住。

二、更合适的任务队列工具推荐 #

用PostgreSQL自制队列虽然可行，但需要自己处理锁、重试、负载均衡等细节，成熟的任务队列工具能大幅简化开发流程，推荐以下选项：

1. RabbitMQ + Celery #

• Celery是Python生态中最常用的任务队列框架，RabbitMQ作为消息中间件提供可靠的消息传递；
• 自动实现负载均衡、任务重试、死信队列等功能，无需手动维护数据库状态；
• 代码简洁，只需定义任务函数，通过Celery的worker自动分发任务到EC2实例。

2. AWS SQS #

• 针对AWS环境的托管式消息队列服务，无需搭建和维护服务器；
• 支持FIFO队列严格保证任务不重复，还能与EC2、Lambda等AWS服务无缝集成；
• 自带消息超时重试、死信队列功能，运维成本极低。

3. Redis + Celery #

• 如果不需要RabbitMQ的高级特性，Redis作为消息中间件更轻量，部署和维护成本更低；
• 适合任务逻辑简单、对消息可靠性要求中等的场景，同样能实现自动负载均衡。

这些工具都比自制PostgreSQL队列更稳定、易用，能让你专注于任务逻辑本身，而非队列的底层维护。

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