检测 RabbitMQ 连接的失效方法

哈喽！大家好，很高兴又见面了，我是golang学习网的一名作者，今天由我给大家带来一篇《检测 RabbitMQ 连接的失效方法》，本文主要会讲到等等知识点，希望大家一起学习进步，也欢迎大家关注、点赞、收藏、转发! 下面就一起来看看吧！

我有一个 go 语言的 rabbitmq 消费者脚本。这是 rabbitmq 教程中的一个简单脚本，使用 streadway/amqp 库。

问题在于，如果 rabbitmq 服务器停止，消费者脚本不会退出；当rabbitmq服务器重新启动时，消费者不再接收消息。

有没有办法检测消费者连接已断开并重新连接，或者至少终止消费者脚本？

我知道图书馆设置了默认的 10 秒。连接的心跳间隔；有可能以某种方式使用它吗？

func main() {
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
    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(
        "test_task_queue", // name
        true,         // durable
        false,        // delete when unused
        false,        // exclusive
        false,        // no-wait
        nil,          // arguments
    )
    failOnError(err, "Failed to declare a queue")

    err = ch.Qos(
        1,     // prefetch count
        0,     // prefetch size
        false, // global
    )
    failOnError(err, "Failed to set QoS")

    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 {
            log.Printf("Received a message: %s", d.Body)
            d.Ack(false)
            dot_count := bytes.Count(d.Body, []byte("."))
            t := time.Duration(dot_count)
            time.Sleep(t * time.Second)
            log.Printf("Done")
        }
    }()

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

解决方案

amqp.connection 具有方法 notifyclose()，该方法返回通道信号传输或协议错误。 所以类似

for {  //reconnection loop
    conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/") //setup
    notify := conn.NotifyClose(make(chan *amqp.Error)) //error channel
...
    ch, err := conn.Channel()
    msgs, err := ch.Consume(
...
    for{  //receive loop
        select {  //check connection
            case err = <-notify:
            //work with error
            break //reconnect
        case d = <- msgs:
            //work with message
        ...
        }
    }
}

今天关于《检测 RabbitMQ 连接的失效方法》的内容就介绍到这里了，是不是学起来一目了然！想要了解更多关于的内容请关注golang学习网公众号！