连接rabbitmq的代码

消费者断线重连：

package main

import (

"log"

"github.com/streadway/amqp"

"database/sql"

_"github.com/go-sql-driver/mysql"

//"time"

)

func insert(db *sql.DB, a []byte) {

stmt, err := db.Prepare("INSERT INTO rabbit (name) VALUES(?)")

stmt.Exec(a)

defer stmt.Close()

if err != nil {

log.Println(err)

return

}

}

func failOnError(err error, msg string) {

if err != nil {

log.Fatalf("%s: %s", msg, err)

}

}

func main() {

//打开mysql

db, err := sql.Open("mysql", "root:0022....hh@tcp(127.0.0.1:3306)/rabbitmq?charset=utf8")

if err != nil {

log.Fatalf("Open database error: %s\n", err)

}

defer db.Close()

err = db.Ping()

if err != nil {

log.Fatal(err)

}

//forever := make(chan bool)

//连接rabbitmq

for {

conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")

if err != nil {

continue

}

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(

"hello", // name

true, // 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

true, // auto-ack

false, // exclusive

false, // no-local

false, // no-wait

nil, // args

)

failOnError(err, "Failed to register a consumer")

// 协程获取消息队列处理结果

go func() {

for d := range msgs {

log.Printf("Received a message: %s", d.Body)

insert(db, d.Body)

}

}()

if conn != nil {

//time.Sleep(50*time.Second)

continue

}

log.Printf(" [*] Waiting for messages. To exit press CTRL+C")

//

}

}

有疑问加站长微信联系(非本文作者))