Créer et utiliser les files d’attentes (queues) avec rabbitMQ

Les files d’attente sont un élément clé de la gestion des communications asynchrones dans les applications distribuées, et RabbitMQ est l’un des systèmes de gestion de files d’attente les plus populaires et les plus fiables. Dans cet article, nous allons configurer et gérer les files d’attente dans RabbitMQ.

RabbitMQ utilise le protocole AMQP (Advanced Message Queuing Protocol) par défaut. Pour ce tutoriel, nous allons utiliser le package pika qui fournit une interface de programmation pour utiliser AMQP et communiquer avec RabbitMQ. Elle vas nous permettre de communiquer avec RabbitMQ de manière efficace en Python.

# Prérequis

• Python >= 3.6
• RabbitMQ (si vous n’avez pas encore installé rabbitMQ veuillez vous referez a l’article précédent)

# Installer pika

Pour utiliser la bibliothèque Pika, vous devez d’abord l’installer. Cela peut être fait en exécutant la commande suivante dans votre console :

pip3 install pika

Une fois l’installation de Pika terminée, vous pouvez commencer à utiliser les fonctions de publication et de réception de messages dans les files d’attentes. Pour cela, vous devez créer deux scripts distincts, nommés “send.py” et “receive.py”, dans lesquels vous pourrez inclure les fonctions correspondantes. Ces scripts vous permettront respectivement d’envoyer des messages à une file d’attente RabbitMQ et de recevoir des messages de cette même file d’attente.

Send.py

#!/usr/bin/env python
import pika

def send():
    try:

        username = "xxxxx"
        password = "xxxxxx"
        host = "xxx.xxx.xxx.xx"
        credentials = pika.PlainCredentials(username, password)
        parameters = pika.ConnectionParameters(host=host,credentials=credentials)
        connection = pika.BlockingConnection(parameters)

        Quser_input = input(" [*] Queue: ")
        Muser_input = input(" [*] Message: ")
        while True:
            channel = connection.channel()

            channel.queue_declare(queue=str(Quser_input))
            channel.basic_publish(exchange='',
                                routing_key=str(Quser_input),
                                body=str(Muser_input))
            print(" [x] Message sent in {} queue.".format(Quser_input))
            Quser_input = input(" [*] Queue: ")
            Muser_input = input(" [*] Message: ")
    except Exception as error:
        print(error)
    except KeyboardInterrupt:
        connection.close()
        print("Bye")
        exit(0)
send()

credentials = pika.PlainCredentials(username, password) permet de définir les informations d’identification pour se connecter au serveur RabbitMQ.

pika.ConnectionParameters(host=host,credentials=credentials) utilise les informations d’identification pour définir les paramètres de connexion. Le paramètre host spécifie l’adresse IP du serveur RabbitMQ. Les paramètres de connexion sont ensuite associés aux informations d’identification en utilisant l’argument credentials=credentials.

connection = pika.BlockingConnection(parameters) permet de créer une connexion bloquante au serveur RabbitMQ en utilisant les paramètres de connexion précédemment définis. La connexion bloquante signifie que le programme sera en attente de la réponse du serveur avant de continuer à exécuter les instructions suivantes.

Les lignes de code suivantes permettent de récupérer via l’entrée standard le nom de la file d’attente et le message à y envoyer : Quser_input = input(” [*] Queue: “) , Muser_input = input(” [*] Message: “).

channel = connection.channel() permet de créer un canal de communication pour envoyer et recevoir des messages via RabbitMQ. Un canal est utilisé pour envoyer et recevoir des messages via une file d’attente spécifique. Cette ligne de code crée un canal de communication pour envoyer des messages à la file d’attente spécifiée par l’utilisateur.
channel.queue_declare(queue=str(Quser_input)) déclare une file d’attente avec le nom spécifié dans la variable Quser_input. Cela signifie qu’une file d’attente avec ce nom sera créée s’il n’existe pas déjà, ou que l’application utilisera la file d’attente existante si elle existe déjà. Cette étape est nécessaire pour s’assurer qu’il y a une file d’attente disponible pour envoyer des messages.
channel.basic_publish(exchange=”, routing_key=str(Quser_input), body=str(Muser_input)) permet d’envoyer un message (body) à la file d’attente spécifiée par l’utilisateur (routing_key=str(Quser_input)).

La boucle while permet de continuer à envoyer des messages à des files d’attente différentes tant que l’utilisateur ne quitte pas le script.

Receive.py

import pika, sys, os

def main():
    credentials = pika.PlainCredentials("altara", "altara01")
    parameters = pika.ConnectionParameters(host="192.168.7.11",credentials=credentials)
    connection = pika.BlockingConnection(parameters)
    channel = connection.channel()
    Quser_input = input("What is the Queue: ")
    channel.queue_declare(queue=str(Quser_input))

    def callback(ch, method, properties, body):
        print(" [x] Received %r" % body.decode('UTF-8'))

    channel.basic_consume(queue=str(Quser_input), on_message_callback=callback, auto_ack=True)

    print(" [*] Waiting for messages in queue {}. To exit press CTRL+C".format(Quser_input))
    channel.start_consuming()
if __name__ == '__main__':
    try:
        main()
    except KeyboardInterrupt:
        print('Interrupted')
        try:
            sys.exit(0)
        except SystemExit:
            os._exit(0)

Ce script ouvre une session de canal (channel) et demande à l’utilisateur d’entrer le nom de la file d’attente (Quser_input) pour laquelle il souhaite consommer les messages.
La fonction callback permet d’imprimer chaque message reçu.
La ligne de code channel.basic_consume() permet d’écouter sur la file d’attente spécifiée et utilise la fonction de rappel (callback) définie pour traiter les messages reçus.
Enfin, la fonction channel.start_consuming() permet de démarrer la consommation des messages.
En cas d’interruption, il affiche un message et termine le script proprement en utilisant sys.exit() ou os._exit().

Conclusion

En conclusion, ces scripts Python utilise la bibliothèque Pika pour établir une connexion à un serveur RabbitMQ, déclarer une file d’attente spécifique et consommer des messages de cette file d’attente. Il permet également à l’utilisateur de spécifier le nom de la file d’attente en utilisant une entrée standard. Ces scripts peuvent être utilisés pour des tâches de surveillance de file d’attente ou de traitement de messages en temps réel. Il montre comment utiliser efficacement Pika pour communiquer avec RabbitMQ en utilisant les protocoles AMQP.