如何在 Go 中为行插入重用单个 Postgres DB 连接？

Question

我正在尝试使用 Go 为从 rabbitmq 接收到的每条新消息插入一行数据到 Postgres 表中，使用到在下面代码的 init 函数中打开的 DB 的单个连接。

代码不是只打开一个连接，而是打开 497 并最大化导致行插入停止...

我尝试使用opening and closing DB connection in Go app和open database connection inside a function这些问题中的信息，说我应该打开一个连接并使用全局数据库来允许主函数将sql语句传递给在init函数中打开的连接。

我以为我已经这样做了，但是正在为每个新行打开一个新连接，因此一旦达到 postgres 连接限制，代码就会停止工作......

我是 Go 新手，编程经验有限，过去两天我一直在尝试理解/解决这个问题，我真的可以通过一些帮助来了解我哪里出了问题......

var db *sql.DB

func init() {
    var err error
    db, err = sql.Open ( "postgres", "postgres://postgres:postgres@SERVER/PORT/DB")
    if err != nil {
        log.Fatal("Invalid DB config:", err)
    }
    if err = db.Ping(); err != nil {
        log.Fatal("DB unreachable:", err)
    }
}

func main() {

// RABBITMQ CONNECTION CODE IS HERE

// EACH MESSAGE RECEIVED IS SPLIT TO LEGEND, STATUS, TIMESTAMP VARIABLES

// VARIABLES ARE PASSED TO sqlSatement    

        sqlStatement := `
        INSERT INTO heartbeat ("Legend", "Status", "TimeStamp")
        VALUES ($1, $2, $3)
`
        // sqlStatement IS THEN PASSED TO db.QueryRow

        db.QueryRow(sqlStatement, Legend, Status, TimeStamp)
    }
}()

<-forever
}

完整代码如下：

package main

import (
    "database/sql"
    "log"
    _ "github.com/lib/pq"

    "github.com/streadway/amqp"
    "strings"
)
var db *sql.DB

func failOnError(err error, msg string) {
    if err != nil {
        log.Fatalf("%s: %s", msg, err)
    }
}

func init() {
    var err error
    db, err = sql.Open ( "postgres", "postgres://postgres:postgres@192.168.1.69:5432/test?sslmode=disable")
    if err != nil {
        log.Fatal("Invalid DB config:", err)
    }
    if err = db.Ping(); err != nil {
        log.Fatal("DB unreachable:", err)
    }
}

func main() {
    conn, err := amqp.Dial("amqp://Admin:Admin@192.168.1.69:50003/")
    failOnError(err, "Failed to connect to RabbitMQ")
    defer conn.Close()

    ch, err := conn.Channel()
    failOnError(err, "Failed to open a channel")
    defer ch.Close()

    q, err := ch.QueueDeclare(
        "HEARTBEAT", // name
        false,       // durable
        false,       // delete when unused
        false,       // exclusive
        false,       // no-wait
        nil,         // arguments
    )
    failOnError(err, "Failed to declare a queue")

    msgs, err := ch.Consume(
        q.Name, // queue
        "",     // consumer
        false,  // auto-ack
        false,  // exclusive
        false,  // no-local
        false,  // no-wait
        nil,    // args
    )
    failOnError(err, "Failed to register a consumer")

    forever := make(chan bool)

    go func() {

        for d := range msgs {
            myString := string(d.Body[:])
            result := strings.Split(myString, ",")
            Legend := result[0]
            Status := result[1]
            TimeStamp := result[2]

            sqlStatement := `
    INSERT INTO heartbeat ("Legend", "Status", "TimeStamp")
    VALUES ($1, $2, $3)
    `
            //
            db.QueryRow(sqlStatement, Legend, Status, TimeStamp)
        }
    }()

    <-forever
}

Answer 1

首先，*sql.DB 不是一个连接，而是一个连接池，它会打开尽可能多的连接，只要它需要和尽可能多的 postgres 服务器允许。只有当池中没有空闲连接可供使用时，它才会打开新连接。

---

所以问题是DB打开的连接没有被释放，为什么？因为您使用的是QueryRow 而没有在返回的*Row 值上调用Scan。

在后台*Row 拥有一个*Rows 实例，该实例可以访问自己的连接，并且在调用Scan 时该连接会自动释放。如果未调用Scan，则不会释放连接，这会导致DB 池在下一次调用QueryRow 时打开一个新连接。因此，由于您没有释放任何连接，DB 会不断打开新连接，直到达到 postgres 设置指定的限制，然后对 QueryRow 的下一次调用挂起，因为它等待连接变为空闲。

因此，如果您不关心输出，则需要使用Exec，或者需要在返回的*Row 上调用Scan。