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

Question

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

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

我已经使用这些问题的信息试图打开和关闭在Go应用程序的数据库连接，并在函数内部开放式数据库连接该说我应该打开一个连接，并使用全球数据库，以便主要功能的SQL语句传递到在打开连接初始化函数。

我以为我已经做到了，但是每行新连接都会打开，因此一旦达到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值。

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

因此，Exec如果您不关心输出，则需要使用，或者需要调用Scanreturn *Row。