How to Detect and Handle Dead RabbitMQ Connections in Go?

Detecting and Handling Dead RabbitMQ Connections in Go

In the provided RabbitMQ consumer script, an interruption in the RabbitMQ server's operation leaves the script running but unresponsive to messages. To address this, it's essential to detect dead connections and either reconnect or terminate the script.

The underlying library, streadway/amqp, offers a solution. The amqp.Connection type has a NotifyClose() method that returns a channel signaling transport or protocol errors. By using this channel, it's possible to detect connection failures.

Here's an updated version of the script that incorporates connection error handling:

import (
    "log"

    "github.com/streadway/amqp"
)

func main() {
    // Connection loop
    for {
        conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
        if err != nil {
            log.Printf("Failed to connect to RabbitMQ: %v", err)
            continue
        }

        notify := conn.NotifyClose(make(chan *amqp.Error))

        ch, err := conn.Channel()
        if err != nil {
            log.Printf("Failed to open a channel: %v", err)
            continue
        }

        q, err := ch.QueueDeclare(
            "test_task_queue",
            true,
            false,
            false,
            false,
            nil,
        )
        if err != nil {
            log.Printf("Failed to declare a queue: %v", err)
            continue
        }

        err = ch.Qos(
            1,
            0,
            false,
        )
        if err != nil {
            log.Printf("Failed to set QoS: %v", err)
            continue
        }

        msgs, err := ch.Consume(
            q.Name,
            "",
            false,
            false,
            false,
            false,
            nil,
        )
        if err != nil {
            log.Printf("Failed to register a consumer: %v", err)
            continue
        }

        // Message receive loop
        for {
            select {
            case err := <-notify:
                // Connection error handling
                log.Printf("Connection error: %v", err)
                break // Reconnect

            case d := <-msgs:
                // Message handling
                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")
            }
        }
    }
}

In this revised script, the connection loop wraps the other operations. Within the message receive loop, a select statement checks both the error channel and incoming messages. If a connection error is detected, it logs the error and triggers reconnection. This mechanism ensures that the script can recover from transient connection issues and resume normal operation.

The above is the detailed content of How to Detect and Handle Dead RabbitMQ Connections in Go?. For more information, please follow other related articles on the PHP Chinese website!