用PGX在Golang进行交易

Question

我目前正在创建一个小小的Go应用程序。现在我正在处理DB部分。我使用的图书馆是这样一个：https://github.com/jackc/pgx

我遇到的问题是，每次我尝试执行数据库读取时，它都会告诉我“conn很忙”。我读过关于使用not池而不是单个连接的文章，但是它仍然不起作用。我做错了什么？

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

在提交事务之前，必须先扫描行。

如果希望事务处理保持在函数中，则可以传递一个接口，该接口也可以在函数中进行扫描。

例如：

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