sql - 使用 PGX 在 Golang 中进行交易

Question

我目前正在创建一个小的 Go App。现在我正在研究数据库部分。我使用的库是这个:https://github.com/jackc/pgx
我遇到的问题是，每次我尝试执行数据库读取时，它都会告诉我“conn is busy”。我阅读了有关使用 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
}

提前致谢。

Answer 1

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

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

例如:

// 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)
}