postgresql - 为什么每笔交易都算作一个客户？

Question

我正在处理一堆文件，然后将结果转储到 PostgreSQL 中。我想同时处理许多 worker ，但不断收到错误消息“pq:抱歉，已经有太多客户了”。只要 worker > 100 左右，这似乎就会发生。 (为简单起见，下面的代码演示了该过程，但我只是在每个表中插入 100 万行，而不是处理文件)。

因为我重复使用相同的 *db，为什么会出现此错误？每笔交易都算作一个客户还是我做错了什么？

package main

import (
    "database/sql"
    "flag"
    "fmt"
    "log"
    "sync"

    "github.com/lib/pq"
)

func process(db *sql.DB, table string) error {
    if _, err := db.Exec(fmt.Sprintf(`DROP TABLE IF EXISTS %v;`, table)); err != nil {
        return err
    }

    col := "age"
    s := fmt.Sprintf(`
        CREATE TABLE %v (
            pk serial PRIMARY KEY,
            %v int NOT NULL
    )`, table, col)

    _, err := db.Exec(s)
    if err != nil {
        return err
    }

    tx, err := db.Begin()
    if err != nil {
        return err
    }

    defer func() {
        if err != nil {
            tx.Rollback()
            return
        }
        err = tx.Commit()
    }()

    stmt, err := tx.Prepare(pq.CopyIn(table, col))
    if err != nil {
        return err
    }

    defer func() {
        err = stmt.Close()
    }()

    for i := 0; i < 1e6; i++ {
        if _, err = stmt.Exec(i); err != nil {
            return err
        }
    }

    return err

}

func main() {
    var u string
    flag.StringVar(&u, "user", "", "user")

    var pass string
    flag.StringVar(&pass, "pass", "", "pass")

    var host string
    flag.StringVar(&host, "host", "", "host")

    var database string
    flag.StringVar(&database, "database", "", "database")

    var workers int
    flag.IntVar(&workers, "workers", 10, "workers")

    flag.Parse()

    db, err := sql.Open("postgres",
        fmt.Sprintf(
            "user=%s password=%s host=%s database=%s sslmode=require",
            u, pass, host, database,
        ),
    )

    if err != nil {
        log.Fatalln(err)
    }

    defer db.Close()
    db.SetMaxIdleConns(0)

    var wg sync.WaitGroup
    ch := make(chan int)

    for i := 0; i < workers; i++ {
        wg.Add(1)
        go func() {
            defer wg.Done()
            for i := range ch {
                table := fmt.Sprintf("_table%d", i)
                log.Println(table)
                if err := process(db, table); err != nil {
                    log.Fatalln(err)
                }
            }
        }()
    }

    for i := 0; i < 300; i++ {
        ch <- i
    }

    close(ch)
    wg.Wait()
}

我知道我可以简单地增加 posgresql 设置，但想了解这个问题:How to increase the max connections in postgres?

Answer 1

Since I am reusing the same *db why am I getting this error?

我怀疑 Postgress 驱动程序正在为您的每个工作人员使用单独的连接，这在大多数情况下都是明智的决定。

Does each transaction count as a client or am I doing something wrong?

在您的情况下，是的，每个事务都算作一个客户端，因为您正在调用 process() 作为 goroutine。您正在创建与工作人员一样多的并发事务。由于您的每个事务都很长，因此所有事务都可能同时使用到数据库的单独连接，因此您达到了限制。

go func() {
        defer wg.Done()
        for i := range ch {
            table := fmt.Sprintf("_table%d", i)
            log.Println(table)
            if err := process(db, table); err != nil {
                log.Fatalln(err)
            }
        }
    }()