Comment détecter et gérer les connexions RabbitMQ mortes dans Go ?

Détection et gestion des connexions RabbitMQ mortes dans Go

Dans le script consommateur RabbitMQ fourni, une interruption du fonctionnement du serveur RabbitMQ laisse le script en cours d'exécution mais ne répond pas aux messages. Pour résoudre ce problème, il est essentiel de détecter les connexions mortes et soit de reconnecter, soit de terminer le script.

La bibliothèque sous-jacente, streadway/amqp, propose une solution. Le type amqp.Connection possède une méthode NotifyClose() qui renvoie un canal signalant des erreurs de transport ou de protocole. En utilisant ce canal, il est possible de détecter les échecs de connexion.

Voici une version mise à jour du script qui intègre la gestion des erreurs de connexion :

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")
            }
        }
    }
}

Dans ce script révisé, la boucle de connexion enveloppe le d'autres opérations. Dans la boucle de réception de messages, une instruction select vérifie à la fois le canal d'erreur et les messages entrants. Si une erreur de connexion est détectée, il enregistre l'erreur et déclenche la reconnexion. Ce mécanisme garantit que le script peut récupérer des problèmes de connexion temporaires et reprendre un fonctionnement normal.

Ce qui précède est le contenu détaillé de. pour plus d'informations, suivez d'autres articles connexes sur le site Web de PHP en chinois!