如何检测死的RabbitMQ连接?

问题描述

我在 Go 中有一个 RabbitMQ 消费者脚本.这是一个来自 RabbitMQ 教程的简单脚本，它使用了 streadway/amqp 库.

I have a RabbitMQ consumer script in Go. This is a simple script from RabbitMQ tutorial that uses streadway/amqp library.

问题是如果RabbitMQ服务器停止，消费者脚本不退出；并且当RabbitMQ服务器重启后，消费者不再接收消息.

The problem is that if the RabbitMQ server is stopped, the consumer script does not exit; and when RabbitMQ server is restarted, the consumer does not receive messages anymore.

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

Is there a way to detect that the consumer connection is dead and reconnect, or at least terminate the consumer script?

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

I know that the library sets a default 10 sec. heartbeat interval for the connection; is it possible to use that someway?

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() 返回通道信号传输或协议错误.所以像

amqp.Connection has method NotifyClose() which return channel signalling a transport or protocol error.So something like

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连接?的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持！