pgx执行tx.Commit()返回conn busy错误的原因及正确实现方案

核心疑问解答 #

• 事务与批量查询可以同时使用：这种组合完全合理，能确保所有批量插入操作具备原子性（要么全部成功，要么全部回滚），适合需要数据一致性的业务场景。
• 批量查询不会自动创建事务：默认情况下，pgx的批量操作会在自动提交模式下执行，即批量中的每个SQL语句各自独立提交（每个语句对应一个单独事务）。如果需要原子性，必须显式开启事务包裹批量操作。

错误原因 #

你遇到的conn busy错误，是因为调用tx.Commit()时，batchRes（批量操作的结果对象）仍持有连接，未释放连接的忙碌状态。代码中defer batchRes.Close()会在函数退出时才执行，而Commit操作发生在这之前，此时连接被batchRes占用，无法完成事务提交。

修正后的代码实现 #

// InsertItems adds items to the table
func (r *Repository) InsertItems(ctx context.Context, values []service.Transaction) error {
    conn, err := r.pool.Acquire(ctx)
    if err != nil {
        return fmt.Errorf("acquire connection: %w", err)
    }
    defer conn.Release()

    tx, err := conn.Begin(ctx)
    if err != nil {
        return fmt.Errorf("starting pgx transaction: %w", err)
    }
    defer func() { _ = tx.Rollback(ctx) }()

    batch := pgx.Batch{}
    // 提前拼接SQL，避免循环内重复格式化
    query := fmt.Sprintf(`INSERT INTO %v (id, date, amount) VALUES ($1, $2, $3)`, r.tableName)
    for _, v := range values {
        batch.Queue(query, v.ID, v.Date, v.Amount)
    }

    batchRes := tx.SendBatch(ctx, &batch)

    cmdTag, err := batchRes.Exec()
    if err != nil {
        // 出错时先关闭batchRes再返回
        _ = batchRes.Close()
        return fmt.Errorf("batch res exec: %w", err)
    }

    // 处理完结果后手动关闭batchRes，释放连接
    if err := batchRes.Close(); err != nil {
        logger.Errorf("closing batch result: %v", err)
    }

    logger.Debugf("inserted rows: %d", cmdTag.RowsAffected())

    // 此时连接已释放，可正常提交事务
    if err := tx.Commit(ctx); err != nil {
        return fmt.Errorf("commiting pgx transaction: %w", err)
    }

    return nil
}

关键修改说明 #

1. 提前拼接SQL语句：将fmt.Sprintf移出循环，避免重复执行字符串格式化，提升代码性能。
2. 手动提前关闭batchRes：在处理完批量结果后立即调用batchRes.Close()，确保连接在Commit前被释放，解决conn busy问题。

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