Parallelizable est une librairie qui permet de paralléliser très facilement une tâche synchrone en la transformant en une tâche asynchrone.

javadoc: http://www.opensides.fr/wp-content/parallelizable/index.html

download lib/source/samples: http://sourceforge.net/projects/parallelizable/files/

project: http://sourceforge.net/projects/parallelizable/

Problème des architectures synchrones

Dans les architectures en couches, chacune des couches délèguent aux couches inférieures un certain nombre de traitements. Dans une architectures synchrone le demandeur (Layer A) attend la réponse de l’exécuteur (Layer B) avant de continuer son travail. Le problème de ce type d’architecture réside dans le fait de bloquer l’appelant le temps d’exécution du traitement:

La librairie Parallelizable propose de permettre à la couche appelante de poursuivre son traitement le temps d’obtenir une réponse de la part de la couche exécutrice.

Comment ça marche ?

L’idée repose sur les architectures SEDA défini dans les année 2000 par Matt Welsh (cf. http://www.eecs.harvard.edu/~mdw/papers/events.pdf).

Le principe est distinguer des « Ã©tapes », basiquement, on peut imaginer ces étapes comme étant les couches de notre architectures, ces étapes étant connectée entre elles par des canaux de communications non-bloquant. De cette façon celui qui publie le message (le demandeur ou producteur) n’attend pas le retour et peut continuer son travail en parallèle. Il sera avertit lorsque la réponse sera prête.

C’est un pattern courant chez les EAI, il est appelé request-reply (cf. http://www.eaipatterns.com/RequestReply.html).

Ici, la couche A publie une requête dans une canal de requête (request) et continue son cycle de vie pendant que la couche B traite en parallèle la requête. Une fois le traitement terminé, la couche B va publier la réponse dans un canal de réponse (reply). Le canal peut être une file d’attente JMS, une queue, une base de donnée, etc…

Les principaux intérêt de ce pattern résident dans la possibilité d’exécuter des tâches en parallèle et d’adapter facilement les producteurs/consommateurs en fonction de l’état des files d’attentes ce qui rend ce type d’architecture hautement scalable.

Conception

Utilisation de la librairie

Exemple d’un appel synchrone

Prenons l’exemple d’une classe SynchronousSample qui fait appel à un service VeryLongLegacyProcess donnant la météo en fonction du jour passé en paramètre. Le temps d’exécution du service est de 5 secondes.

La classe appelante choisi de prendre un parapluie ou une casquette en fonction du temps. Une première implémentation donnerait le code suivant:

La classe SynchronousSample:

public class SynchronousSample {
 VeryLongLegacyProcess veryLongProcess = new VeryLongLegacyProcess();
 public void howDoIDress(EDay day) throws InterruptedException{
 EWeather weather = veryLongProcess.whatTheWeather(day); //1 seconde
 String clothe = "";
 if (EWeather.RAINNY.equals(weather)){
 clothe = "umbrella";
 }else{
 clothe = "cap";
 }
 System.out.println(day+" I will take my "+clothe);
 }
}

La classe Main qui va appeler le traitement:

public class Main {
 public static void main (String[] args){
 System.out.println("current time:"+System.currentTimeMillis());
 LayerA layerA = new LayerA();
 layerA.howDoIDress(EDay.MONDAY);
 layerA.howDoIDress(EDay.TUESDAY);
 layerA.howDoIDress(EDay.WEDNESDAY);
 layerA.howDoIDress(EDay.THURSDAY);
 layerA.howDoIDress(EDay.FRIDAY);

}

Execution

current time:1272183941674
MONDAY I will take my umbrella(current time:1272183946682)
TUESDAY I will take my cap(current time:1272183951682)
WEDNESDAY I will take my umbrella(current time:1272183956683)
THURSDAY I will take my cap(current time:1272183961683)
FRIDAY I will take my umbrella(current time:1272183966683)

job execution: 25 009 ms

=> L’exécution du traitement pour chacun des jours de la semaine prend 25 secondes (5 fois 5 secondes).

Exemple Asynchrone avec Parallelizable

Nous allons maintenant essayer de réduire la fenêtre temporelle de la classe main en parallélisant les appels au service en étendant la classe Parallelizable.

La classe parallelisable:

public class AsynchronousSample extends Parallelizable<EDay, EWeather>{
 VeryLongLegacyProcess veryLongLegacyProcess = new VeryLongLegacyProcess();
 public AsynchronousSample(int nbExecutors) throws InstantiationException,
 IllegalAccessException, IllegalArgumentException,
 SecurityException, InvocationTargetException, NoSuchMethodException {
 super(nbExecutors);
 }

 @Override
 public void callback(Reply<EDay, EWeather> reply) {
 String clothe = "";
 if (EWeather.RAINNY.equals(reply.getOut())){//using results for next process
 clothe = "umbrella";
 }else{
 clothe = "cap";
 }
 System.out.println(reply.getIn()+" I will take my "+clothe);//using input parameters        
 }

 @Override
 public EWeather processRequest(EDay in) throws InterruptedException {
 return veryLongLegacyProcess.whatTheWeather((EDay)in);
 }

}

La classe Main qui va appeler le traitement:

public class Main {
 public static void main (String[] args){

 System.out.println("current time:"+System.currentTimeMillis());

 AsynchronousSample sample = new AsynchronousSample(5);

 sample.request(EDay.MONDAY);

 sample.request(EDay.TUESDAY);

 sample.request(EDay.WEDNESDAY);

 sample.request(EDay.THURSDAY);

 sample.request(EDay.FRIDAY);

}}

Execution

current time:1272184056048
WEDNESDAY I will take my umbrella(current time:1272184061059)
FRIDAY I will take my umbrella(current time:1272184061059)
THURSDAY I will take my cap(current time:1272184061059)
TUESDAY I will take my cap(current time:1272184061059)
MONDAY I will take my umbrella(current time:1272184061059)
job execution: 5 011 ms

La classe Parallelizable va créer un pool de threads exécuteurs, donnée en paramètre au niveau du constructeur (Sample sample = new Sample(5);). Ces threads seront mis en écoute à la première requête publiée dans le canal de requête (avec la méthode request (sample.request(EDay.MONDAY)). Chacun des threads va dépiler les messages envoyés et les traiter dans la méthode processRequest. La réponse sera ensuite publiée dans un canal de réponse, dépilée et envoyée dans la méthode callback.

=> L’exécution du traitement pour chacun des jours de la semaine prend cette fois 5 secondes du fait de leur exécution parallèle.

L’exemple, la librairie et les sources sont publiés sous la licence Apache V2 sur Source Forge https://sourceforge.net/projects/parallelizable/