Connecter RabbitMQ à Talend Studio : Publier et Consommer des Messages en Toute Simplicité

Dans ce tutoriel, vous allez apprendre à configurer Talend Studio pour publier et consommer des messages avec RabbitMQ. Ce guide vous montrera comment utiliser les composants RabbitMQ de Talend pour envoyer des messages à une file d’attente et les lire facilement dans un flux ETL.

Prérequis :

• Avoir installé le Studio Talend.
• Avoir installé RabbitMQ et avoir accès à l’interface de gestion pour créer des files d’attente et des messages.
• Connaissances de base sur le Studio et en intégration de données.

Composants utilisés dans le traitement :

• tRabbitMQConnection : pour établir la connexion entre Talend et le serveur RabbitMQ.
• tRabbitMQClose : pour fermer la connexion entre Talend et le serveur RabbitMQ.
• tRabbitMQOutput : pour envoyer des messages vers une file d’attente RabbitMQ.
• tRabbitMQInput : pour consommer et lire des messages depuis une file d’attente RabbitMQ.
• tWriteJSONField : pour générer un fichier JSON.
• tMap : pour mapper les colonnes du fichier source.
• tLogRow : pour afficher le contenu d’un flux dans la console d’exécution.

Scénario :

Supposons que vous devez intégrer RabbitMQ dans votre flux de données pour envoyer des notifications ou des données de traitement dans une file d’attente. Dans ce tutoriel, nous allons :

1. Établir une connexion avec RabbitMQ depuis Talend.
2. Publier un message dans une file d’attente RabbitMQ à l’aide de tRabbitMQOutput.
3. Consommer le message depuis la file d’attente avec tRabbitMQInput et l’afficher dans la console avec tLogRow.

1. Authentification & connexion au serveur RabbitMQ :

Procédure :

• Dans votre designer ajoutez le composant tRabbitMQConnection et configurez-le en saisissant les paramètres suivants :
• Hostname : Nom/Adresse IP du serveur ou ici en l’occurrence « localhost » pour le test.
• Port : Pour les opérations de messagerie via AMQP, le port 5672 est utilisé.
• Username & Password : Votre nom d’utilisateur & mot de passe créés dans l’interface de gestion en utilisant le port 15672. Pour cet exemple nous utilisons ici « guest »/ »guest » c’est le username et password par default utilisés en local souvent pour les tests. Il possède les droits admin, mais ne peux être utilisé pour des connections à distance.

2. Création d’un message et envoie dans la Queue.

⚠️ Il existe plusieurs façons d’écrire un message dans RabbitMQ : ici, nous allons utilisés les données d’un fichier CSV, mais la source pourrait aussi être une base de données, un service web, une API REST, un fichier XML ou même une entrée manuelle.

Le bloc de votre job devra ressemblé a celui dans la figure ci-dessous :

• Nous allons lire un fichier délimité (ici CSV) avec le tFileInputDelimited et envoyer la donnée sous forme de flux.
• Ensuite est utilisé le tMap pour mapper les données provenant de tFileInputDelimited. Nous ne faisons dans ce job aucune transformation ou filtre de données.
• Vient ensuite la conversion des données mapper en format JSON via un tWriteJSONField. Il prend les colonnes du flux d’entrée et les structure en un objet JSON.
• Pour finir nous utilisons le tRabbitMQOutput pour envoyer les données JSON produites par tWriteJSONField vers une file d’attente RabbitMQ . Utilisez les mêmes paramètres de connexion que sur le composant tRabbitMQConnection.
• Ensuite paramètre : Exchange type sur « Default« , il existe plusieurs type d’exchange entre un Message et une Queue, le plus simple à comprendre est « Default », pour que le message soit délivré à la queue le Binding donc la clé « Routine Key » doit seulement avoir le même nom que la Queue, ici « QueueTest1« . C’est celui que nous utilisons dans l’exemple.

tWarn  : Utilisez le pour afficher un message de log une fois que le sous-job de lecture et d’envoi des données est terminé.

3. Extraction et consommation du message.

Procédure :

• Configurez un composant tRabbitMQInput comme le montre la figure ci-dessous :

• Les paramètres de connexion sont identiques. Le schéma utilisé est le même que celui du tWriteJSONField en sortie.
• Paramètre le nom de la Queue dans laquelle vous voulez consommer un message, ici « QueueTest1« , c’est dans celle là que nous avons envoyé notre message précédemment.
• Ensuite reliez votre tRabbitMQInput à un tLogRow afin d’afficher le contenu de votre message.
• Lorsque vous exécutez un tRabbitMQInput vous allez lire en permanence la queue demandée, pour consommer les messages des qu’ils sont créés. Donc votre Job ne se termine jamais en réalité, sauf si vous ajouté un tDie avec une relation main au tLogRow, alors une erreur fatale se créera et votre message sera tout de même consommé.
• Pour finir, clôturez la connexion avec un tRabbitMQClose, et un tWarn pour être sûr que la connexion est fermée proprement.

Dans Advanced settings du tRabbitMQInput cochez le mode « Auto acknowledge » cela permet de gérer les accusés de réception en automatique, et ne pas avoir à le faire manuellement.

Votre sous- job devra ressembler à celui dans la figure ci-dessous :

Vous savez désormais comment s’authentifier, créer et extraire des messages via RabbitMQ et le protocole AMQP sur Talend Studio.