我正在尝试使用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 }
首先,*sql.DB不是连接而是连接池,它将打开所需数量的连接以及postgres服务器允许的数量。仅当池中没有空闲的可用连接时,它才会打开新连接。
*sql.DB
因此,问题在于数据库打开的连接没有被释放,为什么?因为您使用时QueryRow未调用Scan返回*Row值。
QueryRow
Scan
*Row
引擎盖下*Row有一个*Rows实例,该实例可以访问其自己的连接,并且该连接在Scan被调用时会自动释放。如果Scan未调用,则连接不会释放,这会导致DB池在下一次调用时打开新连接QueryRow。因此,由于您不释放任何连接,因此请DB继续打开新的连接,直到达到postgres设置所指定的限制,然后下一次调用QueryRow挂起,因为它等待连接变为空闲状态。
*Rows
DB
因此,Exec如果您不关心输出,则需要使用,或者需要调用Scanreturn *Row。
Exec