rabbitmq消费者插入mysql数据并实现断线自动重连

连接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") //<-forever } }