使用 PGX 在 Golang 中进行交易答案

【问题标题】：Transaction in Golang with PGX使用 PGX 在 Golang 中进行交易
【发布时间】：2021-11-23 00:51:14
【问题描述】：

我目前正在创建一个小的 Go 应用程序。现在我正在研究数据库部分。我使用的库是这个：https://github.com/jackc/pgx
我遇到的问题是，每次我尝试执行数据库读取时，它都会告诉我我的“conn 很忙”。我读过有关使用 pgxpool 而不是单个连接的信息，但它仍然不起作用。我究竟做错了什么？

func (postgre *PostgreClient) read(query string) (pgx.Row, error) {
    client, err := postgre.client.Acquire(context.TODO())
    transaction, err := client.BeginTx(context.TODO(), pgx.TxOptions{})
    if err != nil {
        return nil, err
    }
    defer transaction.Rollback(context.TODO())

    rows := transaction.QueryRow(context.TODO(), query)
    if err != nil {
        return nil, err
    }
    err = transaction.Commit(context.TODO())
    return rows, err
}

提前致谢。

【问题讨论】：

你从Acquire查看err了吗？
错误为零。如果有帮助，错误本身就是 connLockError。
那么问题代码中的哪一行导致“conn is busy”？这也是您在使用 db 的小 Go 应用程序中唯一的代码吗？有多少个 goroutine 使用read？您的代码中有多少次获取。或者你假设有多少是在运行时执行的？我问是因为问题可能是由于获取太多连接而不释放它们引起的。 read 获得的那一款似乎没有发布。基本上看起来你正在泄漏连接。
提交导致问题。 'QueryRow' 函数执行没有错误。目前这个读取函数只执行一次，因为我只是在测试。
那我唯一能想到的就是结果还没有被扫描。即通常在提交/回滚之前调用row.Scan（我不确定是否需要这样做）。此外，无论查询是否失败，您的代码都会调用 Rollback，这不是好的形式。

标签： sql postgresql go transactions postgis

【解决方案1】：

您必须在提交事务之前扫描该行。

如果您希望事务的处理保留在函数内，您可以传递一个也在函数内进行扫描的接口。

例如：

// implemented by *sql.Row & *sql.Rows
type Row interface {
    Scan(dst ...interface{}) error
}

// implemented by your "models"
type RowScanner interface {
    ScanRow(r Row) error
}

type User struct {
    Id    int
    Email string
}

func (u *User) ScanRow(r Row) error {
    return r.Scan(
        &u.Id,
        &u.Email,
    )
}

func (postgre *PostgreClient) read(query string, rs RowScanner) (err error) {
    conn, err := postgre.client.Acquire(context.TODO())
    if err != nil {
        return err
    }
    defer conn.Release()
    
    tx, err := conn.BeginTx(context.TODO(), pgx.TxOptions{})
    if err != nil {
        return err
    }
    defer func() {
        if err != nil {
            tx.Rollback(context.TODO())
        } else {
            tx.Commit(context.TODO())
        }
    }()

    row := tx.QueryRow(context.TODO(), query)
    if err != nil {
        return nil, err
    }
    return rs.ScanRow(row) 
}

u := new(User)
if err := pg.read("select id, email from users limit 1", u); err != nil {
    panic(err)
}

用于扫描模型列表：

type UserList []*User

func (ul *UserList) ScanRow(r Row) error {
    u := new(User)
    if err := u.ScanRow(r); err != nil {
        return err
    }

    *ul = append(*ul, u)
    return nil
}

func (postgre *PostgreClient) list(query string, rs RowScanner) (err error) {
    conn, err := postgre.client.Acquire(context.TODO())
    if err != nil {
        return err
    }
    defer conn.Release()
    
    tx, err := conn.BeginTx(context.TODO(), pgx.TxOptions{})
    if err != nil {
        return err
    }
    defer func() {
        if err != nil {
            tx.Rollback(context.TODO())
        } else {
            tx.Commit(context.TODO())
        }
    }()

    rows, err := tx.Query(context.TODO(), query)
    if err != nil {
        return err
    }
    defer rows.Close()
    
    for rows.Next() {
        if err := rs.ScanRow(rows); err != nil {
            return err
        }
    }
    return rows.Err()
}

ul := new(UserList)
if err := pg.list("select id, email from users", ul); err != nil {
    panic(err)
}

【讨论】：