基于PostgreSQL的Java应用任务队列重复执行问题求助

这个问题太常见了——多工作线程抢任务导致重复执行，本质就是没有在任务获取阶段做排他性的行级锁定。结合PostgreSQL的特性，给你几个落地性强的解决方案，按推荐程度排序：

这是PostgreSQL 9.5+支持的特性，完美适配“抢任务”场景。它的核心逻辑是：查询符合调度时间条件的任务时，自动锁定选中的行，并且跳过已经被其他事务锁定的行。这样每个工作线程拿到的都是唯一的、未被处理的任务，从根源避免重复执行。

具体实现（Java+JDBC示例） #

try (Connection conn = dataSource.getConnection()) {
    conn.setAutoCommit(false); // 必须开启事务，否则锁会立即释放

    String sql = "SELECT id, task_json FROM task_queue " +
                 "WHERE schedule_time <= NOW() " +
                 "ORDER BY schedule_time ASC " +
                 "LIMIT 1 " +
                 "FOR UPDATE SKIP LOCKED"; // 关键锁定语句

    try (PreparedStatement stmt = conn.prepareStatement(sql)) {
        ResultSet rs = stmt.executeQuery();
        if (rs.next()) {
            long taskId = rs.getLong("id");
            String taskJson = rs.getString("task_json");

            // 执行你的任务逻辑
            executeTask(taskJson);

            // 执行成功后删除任务（或更新状态保留历史）
            String deleteSql = "DELETE FROM task_queue WHERE id = ?";
            try (PreparedStatement deleteStmt = conn.prepareStatement(deleteSql)) {
                deleteStmt.setLong(1, taskId);
                deleteStmt.executeUpdate();
            }

            conn.commit(); // 提交事务，释放锁
        } else {
            conn.commit(); // 没有待执行任务，直接提交
        }
    } catch (Exception e) {
        conn.rollback(); // 执行失败回滚，释放锁
        throw e;
    }
}

优点 #

• 简单直接，数据库层面保证排他性，无需额外字段
• 性能优异，跳过锁定行的逻辑由数据库优化，避免线程无效等待
• 天然支持负载均衡，多个线程自动分配不同任务

注意事项 #

• 必须在事务中执行，否则锁会立即释放
• 任务执行时间不宜过长，避免占用数据库锁资源（可设置事务超时）

如果你的PostgreSQL版本低于9.5（不支持SKIP LOCKED），可以通过给任务表添加状态字段来控制任务所有权。

步骤： #

1. 给任务表新增status字段（比如VARCHAR(20)，可选值：PENDING, PROCESSING, COMPLETED, FAILED），以及version字段（乐观锁用，INT默认0）
2. 工作线程先尝试将PENDING状态的任务更新为PROCESSING，更新成功的线程拥有任务处理权

乐观锁实现（Java+Spring Data JPA示例） #

// 实体类定义
@Entity
@Table(name = "task_queue")
public class Task {
    @Id
    private Long id;
    @Column(columnDefinition = "JSONB")
    private String taskJson;
    private LocalDateTime scheduleTime;
    private String status = "PENDING";
    @Version // Spring Data JPA乐观锁注解
    private Integer version;
    // getter/setter省略
}

// 业务逻辑代码
@Transactional
public Optional<Task> acquireTask() {
    // 查询符合条件的待处理任务
    Task task = taskRepository.findFirstByStatusAndScheduleTimeLessThanEqual("PENDING", LocalDateTime.now());
    if (task == null) {
        return Optional.empty();
    }

    try {
        // 更新状态为PROCESSING，乐观锁会自动校验version
        task.setStatus("PROCESSING");
        taskRepository.save(task);
        return Optional.of(task);
    } catch (OptimisticLockingFailureException e) {
        // 说明其他线程已抢占该任务，直接放弃
        return Optional.empty();
    }
}

// 任务执行完成后
@Transactional
public void completeTask(Long taskId) {
    taskRepository.deleteById(taskId);
    // 若需保留历史，可改为更新状态：taskRepository.updateStatusById("COMPLETED", taskId);
}

优点 #

• 兼容低版本PostgreSQL
• 可保留任务执行历史（不删除仅更新状态），便于排查问题
• 状态字段便于监控任务生命周期

缺点 #

• 乐观锁可能导致“空转”（多线程抢同一任务，仅一个成功），高并发下略有性能损耗
• 需要额外维护状态和version字段

如果需要更实时的任务调度，可结合PostgreSQL的LISTEN/NOTIFY机制，避免轮询带来的数据库压力。

核心逻辑： #

1. 插入新任务时，通过触发器触发pg_notify发送通知
2. 工作线程监听指定通知频道，收到通知后再去获取任务（结合方案1的SELECT ... FOR UPDATE SKIP LOCKED）
3. 保留轮询逻辑作为 fallback，防止通知丢失

PostgreSQL触发器示例 #

CREATE OR REPLACE FUNCTION notify_new_task()
RETURNS TRIGGER AS $$
BEGIN
    PERFORM pg_notify('new_task_channel', NEW.id::TEXT);
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE TRIGGER trigger_new_task
AFTER INSERT ON task_queue
FOR EACH ROW
EXECUTE FUNCTION notify_new_task();

Java监听代码示例 #

// 启动独立线程监听通知
new Thread(() -> {
    try (Connection conn = dataSource.getConnection()) {
        try (Statement stmt = conn.createStatement()) {
            stmt.execute("LISTEN new_task_channel");
        }

        while (!Thread.currentThread().isInterrupted()) {
            // 等待通知，超时时间设为0表示一直等待
            conn.waitForNotification(0);
            // 收到通知后执行任务获取逻辑（复用方案1的代码）
            acquireAndExecuteTask();
        }
    } catch (Exception e) {
        e.printStackTrace();
    }
}).start();

优点 #

• 减少轮询带来的数据库负载
• 任务触发更实时

缺点 #

• 实现复杂度高，需处理通知丢失、线程重启等异常场景
• 依赖PostgreSQL特性，迁移到其他数据库成本高

• 任务超时处理：若工作线程执行任务时崩溃，任务可能一直处于锁定/处理中状态。可定时扫描：
  • 方案1：查询被锁定超过N分钟的任务，通过pg_locks系统表强制解锁
  • 方案2：将PROCESSING状态且超过超时时间的任务重置为PENDING
• 批量获取任务：任务量大时，可将LIMIT 1改为LIMIT 10，一次获取多个任务，减少数据库查询次数
• 索引优化：给schedule_time和status字段添加联合索引，加快任务查询速度

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