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

Question

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

代码不是只打开一个连接，而是打开 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 服务器允许的尽可能多的连接。它仅在池中没有空闲连接可供使用时才打开新连接。

---

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

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

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