Golang implémente la surveillance RabbitMQ

Nous savons que la file d'attente de messages est un modèle architectural couramment utilisé pour résoudre des problèmes tels que le traitement asynchrone et la distribution des tâches, et RabbitMQ est actuellement l'un des middlewares de messages les plus largement utilisés. Dans des applications pratiques, nous devrons peut-être utiliser Golang pour implémenter la surveillance RabbitMQ. Cet article explique comment utiliser Golang pour implémenter la surveillance RabbitMQ.

Préparation

Avant de commencer, vous devez vous assurer que RabbitMQ a été installé. Puisque RabbitMQ dépend d'Erlang, Erlang doit également être installé.

Une fois l'installation terminée, nous devons installer le package tiers Golang. Parmi eux, le package AMQP est essentiel, qui nous permet de connecter et d'exploiter facilement RabbitMQ.

allez chercher github.com/streadway/amqp

Implémentation du code

Tout d'abord, nous devons nous connecter à RabbitMQ. Une fois la connexion réussie, nous devons déclarer un échange nommé « test » et taper « fanout ». Exchange est une partie importante du routage des messages dans RabbitMQ. Il est responsable de la réception des messages et de leur distribution dans les files d'attente. Dans ce cas, nous déclarerons un échange appelé « test » et définirons son type sur « fanout », ce qui signifie qu'il diffusera des messages à toutes les files d'attente qui y sont abonnés.

conn, err := amqp.Dial("amqp://guest:guest@localhost:5672/")
failOnError(err, "Échec de la connexion à RabbitMQ")
refer conn.Close()

ch, err := conn.Channel()
failOnError(err, "Échec de l'ouverture d'un canal")
defer ch.Close()

err = ch.ExchangeDeclare(

"test",   // name
"fanout", // type
true,     // durable
false,    // auto-deleted
false,    // internal
false,    // no-wait
nil,      // arguments

)
failOnError(err, "Échec de la déclaration d'un Exchange")

Ensuite, nous devons créer une nouvelle file d'attente non persistante avec un nom généré automatiquement. Ici, nous utiliserons les noms des files d'attente pour les lier à l'échange "test" que nous venons de déclarer.

q, err := ch.QueueDeclare(

"",    // name
false, // durable
false, // delete when unused
true,  // exclusive
false, // no-wait
nil,   // arguments

)
failOnError(err, "Échec de la déclaration d'une file d'attente")

err = ch.QueueBind(

q.Name, // queue name
"",     // routing key
"test", // exchange
false,
nil,

)
failOnError(err, "Échec de la liaison d'un file d'attente")

Maintenant, RabbitMQ est prêt et nous pouvons commencer à écouter ses messages. Nous pouvons utiliser la fonction Consume pour implémenter l'écoute des messages, ce qui nous permet de recevoir en continu les messages de la file d'attente et de les traiter.

msgs, err := ch.Consume(

q.Name, // queue name
"",     // consumer
true,   // auto-ack
false,  // exclusive
false,  // no-local
false,  // no-wait
nil,    // args

)
failOnError(err, "Impossible d'enregistrer un consommateur")

for msg := range msgs {

log.Printf("Received a message: %s", msg.Body)

}

Dans le code ci-dessus, nous utilisons La méthode ch.Consume() est utilisée pour écouter les messages dans la file d'attente spécifiée et afficher le contenu du message en imprimant les journaux. Il est à noter que nous utilisons une boucle infinie pour déployer l'écoute des messages, ce qui signifie que nous continuerons à écouter la file d'attente jusqu'à ce que le programme soit arrêté ou qu'une erreur se produise.

Le code complet est le suivant :

package main

import (

"log"

"github.com/streadway/amqp"

)

func failOnError(err error, msg string) {

if err != nil {
    log.Fatalf("%s: %s", msg, err)
}

}

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()

err = ch.ExchangeDeclare(
    "test",   // name
    "fanout", // type
    true,     // durable
    false,    // auto-deleted
    false,    // internal
    false,    // no-wait
    nil,      // arguments
)
failOnError(err, "Failed to declare an exchange")

q, err := ch.QueueDeclare(
    "",    // name
    false, // durable
    false, // delete when unused
    true,  // exclusive
    false, // no-wait
    nil,   // arguments
)
failOnError(err, "Failed to declare a queue")

err = ch.QueueBind(
    q.Name, // queue name
    "",     // routing key
    "test", // exchange
    false,
    nil,
)
failOnError(err, "Failed to bind a queue")

msgs, err := ch.Consume(
    q.Name, // queue name
    "",     // consumer
    true,   // auto-ack
    false,  // exclusive
    false,  // no-local
    false,  // no-wait
    nil,    // args
)
failOnError(err, "Failed to register a consumer")

for msg := range msgs {
    log.Printf("Received a message: %s", msg.Body)
}

}

Résumé

Cet article explique comment utiliser Golang pour implémenter la surveillance Rabbitmq. Tout d'abord, nous devons nous connecter à Rabbitmq, déclarer un échange, créer une file d'attente et lier la file d'attente à l'échange, et enfin utiliser des consommateurs pour surveiller les messages dans la file d'attente. J'espère que cet article pourra être utile aux développeurs qui utilisent Golang pour le développement de Rabbitmq.

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!