Architecture multithread pour une application d'alimentation

Question

Ceci est mon premier article ici, donc excuses si cela n'est pas bien structuré.

Nous avons été chargés de concevoir un outil qui:

• Lire un fichier (d'ID de compte), format CSV
• Télécharger le fichier de données de compte à partir du Web pour chaque compte (par Id) (API REST) ​​
• passer le fichier à un convertisseur qui produira un rapport (de prévisions financières, etc.) [~ 20ms]
• Si le seuil de prédiction est dans les limites, exécutez un analyseur pour analyser les données [400ms]
• Générateur e un rapport pour l'analyse ci-dessus [80ms]
• Télécharger tous les fichiers générés sur le Web (API REST) ​​

Maintenant, tous les points individuels sont relativement faciles à faire. Je suis intéressé à trouver la meilleure façon d'architecturer quelque chose pour gérer cela et de le faire rapidement & efficacement sur notre matériel.

Nous devons traiter environ 2 millions de comptes. Les crochets donnent une idée de la durée moyenne de chaque processus. Je voudrais utiliser le maximum de ressources disponibles sur la machine - 24 processeurs Xeon de base. Ce n'est pas un processus intensif en mémoire.

L'utilisation de la TPL et la création de chacune de ces tâches serait-elle une bonne idée? Chacun doit se faire de manière séquentielle, mais plusieurs peuvent être faits en même temps. Malheureusement, les parseurs ne sont pas multi-threading et nous n'avons pas la source (c'est essentiellement une boîte noire pour nous).

Mes pensées étaient quelque chose comme ça - On suppose que nous utilisons TPL:

• données de compte de la charge (essentiellement d'importation CSV ou SQL SELECT)
• Pour chaque compte (Id):
  • Télécharger le fichier de données pour chaque compte
  • ContinuerAvec l'utilisation du fichier de données, envoyer au convertisseur
  • ContinuerAvec le seuil de vérification, envoyer au parseur
  • ContinueWith Générer un rapport
  • ContinueWith Télécharger sorties

Est-ce que son possible ou que je ne comprenais pas correctement? Serait-il préférable de décomposer les étapes d'une manière différente? Je ne suis pas sûr de la façon de gérer les problèmes avec les exceptions de lancement de l'analyseur (c'est très difficile) ou quand nous obtenons des échecs de téléchargement.

Tout cela va être dans un travail planifié qui sera exécuté après les heures comme une application de la console.

Answer 1

Je pense sur l'utilisation de certains type de messagebus. Vous pouvez donc séparer les étapes et si vous ne pouvez pas travailler (par exemple, parce que le service REST n'est pas accessible pendant un certain temps), vous pouvez stocker le message pour les traiter plus tard. En fonction de ce que vous utilisez comme messagebus, vous pouvez introduire des threads avec celui-ci. À mon avis, vous pourriez mieux concevoir des flux de travail, gérer des états exceptionnels et ainsi de suite, si vous avez une abstraction de plus haut niveau comme un bus de service.

De même, les pièces peuvent fonctionner de manière indépendante et ne se bloquent pas les unes les autres.

Un moyen facile pourrait être d'utiliser servicestack messaging avec Redis ServiceBus.

Certains avantages cités de là:

• conception à base de messages permet de faciliter la parallélisation et l'introspection des calculs

• messages DLQ peuvent introspecté, fixes et plus tard rejoué après les mises à jour du serveur et Rejoindre le flux de travail de message normal

Answer 2

Je pense que la manière facile de commencer avec de multiples fil dans votre cas, sera de mettre l'opération pour chaque identifiant de compte dans un thread (ou mieux, dans un ThreadPool). De la manière proposée ci-dessous, je pense que vous n'aurez pas besoin de contrôler les opérations inter-thread.

Quelque chose comme ceci pour mettre les données sur la file d'attente de pool de threads :

var accountIds = new List<int>(); 
foreach (var accountId in accountIds) 
{ 
    ThreadPool.QueueUserWorkItem(ProcessAccount, accountId); 
} 

Et ceci est la fonction que vous traiterez chaque compte:

public static void ProcessAccount(object accountId) 
{ 
    // Download the data file for this account 
    // ContinueWith using the data file, send to the converter 
    // ContinueWith check threshold, send to parser 
    // ContinueWith Generate Report 
    // ContinueWith Upload outputs 
}