I'm trying to use Go to insert a row of data into a Postgres table for each new message received from rabbitmq, using a single connection to the DB which is opened in the init function of the code below.
Rather than opening just one connection, the code opens 497 and maxes out which causes the row inserts to stop...
I have tried using the info in these questions opening and closing DB connection in Go app and open database connection inside a function which say I should open one connection and use global db to allow the main function to pass the sql statement to the connection opened in the init function.
I thought I had done this, however a new connection is being opened for each new row so the code stops working once the postgres connection limit is reached...
I am new to Go and have limited programming experience, I have been trying to understand/resolve this issue for the past two days and I could really do with some help understanding where I am going wrong with this...
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
}
Full code is shown below:
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:[email protected]: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:[email protected]: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
}
First off, *sql.DB
is not a connection but a pool of connections, it will open as many connection as it needs to and as many as the postgres server allows. It only opens new connections when there is no idle one in the pool ready for use.
So the issue is that the connections that DB opens aren't being released, why? Because you're using QueryRow
without calling Scan
on the returned *Row
value.
Under the hood *Row
holds a *Rows
instance which has access to its own connection and that connection is released automatically when Scan
is called. If Scan
is not called then the connection is not released which causes the DB
pool to open a new connection on the next call to QueryRow
. So since you're not releasing any connections DB
keeps opening new ones until it hits the limit specified by the postgres settings and then the next call to QueryRow
hangs because it waits for a connection to become idle.
So you either need to use Exec
if you don't care about the output, or you need to call Scan
on the returned *Row
.
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With