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不打电话Scan在返回*Row值(value)。

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

所以你要么需要使用Exec如果您不关心输出，或者您需要调用 Scan在返回*Row .