Utilisation du multi-thread sur Talend

Les threads (ou processus en français) sont un moyen de répartir la charge d'un traitement sur plusieurs unités de calcul. Talend offre la possibilité de lancer des sous-jobs en multi-thread, afin de paralléliser les exécutions. Voici une proposition de mise en place d'un tel traitement.

Création du job à multi-threader

Il faut tout d'abord isoler le traitement à lancer en thread, afin qu'il puisse être exécuté de façon indépendante. Créez un job et veillez notamment à bien gérer la concurrence entre les futurs threads, en n'écrivant pas dans des ressources partagées (fichiers, mêmes lignes d'une table en base de données...).

En cas de connexion à une base de données, le composant de connexion ne doit pas utiliser de connexion partagée (décocher la case "Utiliser ou enregistrer une connexion partagée à une base de données).

Il peut être intéressant que le thread lancé sache quel thread il représente. Aussi, vous pouvez ajouter deux paramètres dans le contexte indiquant le numéro de thread et le nombre de threads :

Ainsi, les logs pourront afficher les informations du thread en cours comme ceci :

Pour garder en tête que le traitement sera lancé en multi-thread, nommez le job avec un nom explicite ("threadImport" par exemple).

Appel du job en multi-threads

Créez ensuite un deuxième job. Placez sur le projet un tRunJob appelant le premier job créé et répétez cela autant de fois que vous souhaitez créer de threads. Dans l'exemple ci-dessous, 3 threads seront générés à l'exécution:

Sur les paramètres du tRunJob, vous pouvez affecter une valeur aux deux variables créées précédemment (num_thread et nb_threads) :

Dans les paramètres du Job parent, allez dans Extra et cochez Exécution multi-thread. Cette case à cocher précise que tous les sous-jobs contenus sur le traitement seront lancés en mode multi-thread, et non pas consécutivement.

Sauvez votre job et nommez-le avec un nom explicite, indiquant qu'il effectue du multi-thread (comme "multithreadImport").

Testez ensuite le traitement, vous devez avoir un affichage proche de ceci, dans un ordre d'exécution variable :

-- Lancement du thread 1/3 --
-- Lancement du thread 2/3 --
-- Lancement du thread 3/3 --

Traiter une liste en multi-thread

Il est possible de partager le traitement d'une liste entre les différents threads créés :

• Créez une routine "Statique.java" incluant le code Java suivant :
  
  public static List<String> liste = new ArrayList<>();
  La liste ainsi créée sera visible et partagée entre tous les jobs.
• Créez un nouveau job et remplissez la liste depuis un fichier ou une requête (en créant un tDBInput1 relié à un tJavaRow par exemple)
• Le tJavaRow contiendra le code (s'il s'agit d'une liste de clients) :
  
  Statique.liste.add (input_row.client);
• Ajoutez un tRunJob pour appeler le job qui effectue du multi-thread.
• Sur le job en fin de chaîne (celui-lancé en multi-thread), filtrez la liste pour n'en traiter qu'une partie. Vous allez créer ceci :

• Ajoutez un tJavaFlex qui va parcourir la liste des données. Mettez le "for" dans la partie "Code initial" et l'accolade de fin dans la partie "Code final".
  
  for (String client : Statique.liste) {
      row1.client=client;
  }
• Placez un tMap et ajoutez un champ "numero_ligne" permettant de compter les numéros de lignes :
  
  Affectez sur numero_ligne le code suivant, utilisant le numéro du thread afin de ne pas avoir de concurrence :
  
  Numeric.sequence("ordre"+context.num_thread,1,1)
• Placez un tFilterRow pour filtrer les lignes concernées par le job et spécifiez la condition suivante, permettant de séparer les lignes selon le numéro et le nombre de threads maximum:
  
  input_row.numero_ligne % context.nb_threads == (context.num_thread-1)
• Stockez le tout dans un tHashOutput qui contiendra les données filtrées.
• Lancez le job principal pour vérifier que tout se passe bien.