Go+PostgreSQL订单调度器重复处理与崩溃重执行问题求助

针对你遇到的订单重复处理（崩溃重启/多实例竞争）问题，结合Go和PostgreSQL技术栈，以下是工业界常用的标准解决方案：

一、核心思路：原子性任务认领 + 幂等性保障 #

解决问题的关键是确保同一订单只能被一个实例认领处理，同时即使重复处理也不会产生业务副作用。

二、具体方案 #

方案1：预认领模式（推荐，适合API处理耗时场景） #

通过原子性更新订单状态为PROCESSING来锁定任务，避免多实例竞争，同时处理崩溃后的超时任务。

步骤：

1. 原子认领订单：开启事务，批量将PENDING订单转为PROCESSING，并返回订单详情（用PostgreSQL的UPDATE ... RETURNING实现原子操作）

BEGIN;
UPDATE orders
SET status = 'PROCESSING', updated_at = NOW()
WHERE status = 'PENDING'
LIMIT 200
RETURNING id, external_api_id, amount; -- 返回处理需要的字段
COMMIT;

2. 调用外部API处理：针对认领的订单调用API，注意将订单ID作为幂等键传入API（比如放在请求头或参数中），确保同一订单多次调用仅执行一次。
3. 更新处理结果：批量更新成功/失败订单的状态

-- 更新成功订单
UPDATE orders SET status = 'COMPLETED' WHERE id = ANY(<success_id_list>);
-- 更新失败订单
UPDATE orders SET status = 'FAILED' WHERE id = ANY(<failure_id_list>);

4. 超时任务重置：添加定时任务，将长时间处于PROCESSING状态的订单重置为PENDING（避免崩溃后任务卡住）

UPDATE orders
SET status = 'PENDING'
WHERE status = 'PROCESSING' AND updated_at < NOW() - INTERVAL '5 minutes'; -- 超时时间根据业务调整

Go代码示例（使用database/sql和lib/pq）：

type Order struct {
    ID             int
    ExternalAPIID  string
    Amount         float64
}

// 认领订单
func claimOrders(db *sql.DB) ([]Order, error) {
    tx, err := db.Begin()
    if err != nil {
        return nil, err
    }
    defer tx.Rollback()

    rows, err := tx.Query(`
        UPDATE orders
        SET status = 'PROCESSING', updated_at = NOW()
        WHERE status = 'PENDING'
        LIMIT 200
        RETURNING id, external_api_id, amount
    `)
    if err != nil {
        return nil, err
    }
    defer rows.Close()

    var orders []Order
    for rows.Next() {
        var o Order
        if err := rows.Scan(&o.ID, &o.ExternalAPIID, &o.Amount); err != nil {
            return nil, err
        }
        orders = append(orders, o)
    }
    if err := rows.Err(); err != nil {
        return nil, err
    }

    if err := tx.Commit(); err != nil {
        return nil, err
    }
    return orders, nil
}

// 处理订单并调用API
func processOrders(orders []Order, apiClient *ExternalAPIClient) ([]int, []int) {
    var successIDs, failureIDs []int
    for _, o := range orders {
        // 传入订单ID作为幂等键
        success, err := apiClient.Process(o.ExternalAPIID, o.ID)
        if err != nil || !success {
            failureIDs = append(failureIDs, o.ID)
            continue
        }
        successIDs = append(successIDs, o.ID)
    }
    return successIDs, failureIDs
}

// 更新订单状态
func updateOrderStatuses(db *sql.DB, successIDs, failureIDs []int) error {
    tx, err := db.Begin()
    if err != nil {
        return err
    }
    defer tx.Rollback()

    if len(successIDs) > 0 {
        _, err := tx.Exec(`
            UPDATE orders SET status = 'COMPLETED' WHERE id = ANY($1)
        `, pq.Array(successIDs))
        if err != nil {
            return err
        }
    }

    if len(failureIDs) > 0 {
        _, err := tx.Exec(`
            UPDATE orders SET status = 'FAILED' WHERE id = ANY($1)
        `, pq.Array(failureIDs))
        if err != nil {
            return err
        }
    }

    return tx.Commit()
}

// 定时重置超时订单
func resetTimedOutOrders(db *sql.DB) error {
    _, err := db.Exec(`
        UPDATE orders
        SET status = 'PENDING'
        WHERE status = 'PROCESSING' AND updated_at < NOW() - INTERVAL '5 minutes'
    `)
    return err
}

方案2：行级锁+跳过已锁定订单（适合API处理快速场景） #

利用PostgreSQL的SELECT FOR UPDATE SKIP LOCKED语法，查询时直接锁定订单，跳过已被其他实例锁定的订单，确保同一订单仅被一个实例获取。

步骤：

1. 查询并锁定订单：开启事务，查询PENDING订单并加排他锁，跳过已锁定的记录

BEGIN;
SELECT id, external_api_id, amount
FROM orders
WHERE status = 'PENDING'
LIMIT 200
FOR UPDATE SKIP LOCKED;

2. 调用API处理订单：同样需要传入订单ID作为幂等键
3. 更新状态并提交事务：处理完成后更新订单状态，提交事务释放锁

注意事项：

• 事务持有锁的时间等于API处理时间，若API耗时较长，会降低系统并发度，因此仅适合API处理快速的场景。

三、必加保障：API幂等性 #

无论采用哪种锁机制，都可能因网络波动、服务崩溃等原因导致订单被重复处理，因此外部API必须支持幂等性：

• 以订单ID作为幂等标识，API收到同一订单ID的请求时，直接返回之前的处理结果，不重复执行业务逻辑。
• 可选：在订单表中添加version字段（乐观锁），更新状态时校验版本号，避免重复更新：

UPDATE orders
SET status = 'COMPLETED', version = version + 1
WHERE id = <order_id> AND version = <current_version>;

若更新行数为0，说明该订单已被处理，直接跳过。

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