Discussion :

[mrboliboli]

Bonjour,
J'ai suivi un tuto sur le framwork Executor, mais il reste un petit point d'ombre sur le CompletionService.
Je souhaiterai lancer de nouveaux thread des qu'un future est disponible vie le take() du CompletionService.
Je ne vois pas du tout comment effectuer cela.

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
 
//...
	private List<Future<Administre>> futuresAdm = null;
	private List<Callable<Administre>> taskAdm = null;
	private List<Future<Administre>> futuresData = null;
	private List<Callable<Administre>> taskAdmData = null;
//...
public void start(){
		// lancement du pool de recuperation des administres
		try {
			// on recupere une liste de nni a traiter pour les ajouter a la liste des taches (thread producteur)
//			daoInfx = new DaoAdministreGds();
			daoInfx = new DaoBouchAdministreGds();
			for (Administre adm : daoInfx.findAll(false)){
				taskAdm.add(new ImportAdministreCallable(adm));
			}
			// declaration de l'executor en pool de thread limité au nombre de cpu disponible sur la machine
			ExecutorService executorAdm = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
			ExecutorService executorData = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
			// declaration du Completion service pour traitement des resultats des que disponibles
			CompletionService<Administre> completionServiceAdm = new ExecutorCompletionService<>(executorAdm);
			CompletionService<Administre> completionServiceData = new ExecutorCompletionService<>(executorData);
			// ajout des taches a traiter a la liste de Futurs<T>
			for (Callable<Administre> c : taskAdm){
				futuresAdm.add(completionServiceAdm.submit(c));
			}
//...

voila ce que je souhaiterai faire des qu'un objet future est disponible, mais comment?

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
 
				task = new ImportDataFromAdmCallable(completionServiceAdm.take().get());
				taskAdmData.add(task);
				futuresData.add(completionServiceData.submit(task));

[mrboliboli]

Re bonjour,
J'ai mis une simple boucle while.
Seulement, je ne trouve pas de condition pour sortir de la boucle.
Tous les traitements s’exécutent correctement, mais le programme ne termine jamais.
J'ai essayé avec la méthode .isTerminated() de L'ExecutorService, mai cela ne semble pas fonctionner.
Si quelqu'un a un idée...

pour info :

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
while(!executorAdm.isTerminated()){
	// creation des taches de recuperation des donnés des qu'un adm est créé 
	task = new ImportDataFromAdmCallable(completionServiceAdm.take().get());
	taskAdmData.add(task);
	futuresData.add(completionServiceData.submit(task));
}

[mrboliboli]

J'ai trouvé une piste.
http://www.developpez.net/forums/d61...s/#post3613506

Cela fonctionne correctement si j'ai 30 administrés a traiter, par contre si je passe a 300 le programme ne s’arrête pas..

Code :

Sélectionner tout - Visualiser dans une fenêtre à part

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
			daoInfx = new DaoBouchAdministreGds(300);
			taskAdm = new ArrayList<>();
			for (Administre adm : daoInfx.findAll(false)){
				taskAdm.add(new ImportAdministreCallable(adm));
			}
			// declaration de l'executor en pool de thread limité au nombre de cpu disponible sur la machine
			ExecutorService executorAdm = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
			ExecutorService executorData = Executors.newFixedThreadPool(Runtime.getRuntime().availableProcessors());
			if (log.isDebugEnabled()) log.debug(" pour " + Runtime.getRuntime().availableProcessors() + " Threads" );
			// declaration du Completion service pour traitement des resultats des que disponibles
			CompletionService<Administre> completionServiceAdm = new ExecutorCompletionService<>(executorAdm);
			CompletionService<Administre> completionServiceData = new ExecutorCompletionService<>(executorData);
			// ajout des taches a traiter a la liste de Futurs<T>
			for (Callable<Administre> c : taskAdm){
				completionServiceAdm.submit(c);
			}
			executorAdm.shutdown();
			// lancement des threads fils.
			ImportDataFromAdmCallable task = null;
			while(!executorAdm.isTerminated()){
				// creation des taches de recuperation des données des qu'un adm est créé
				Future<Administre> futAdm = completionServiceAdm.take();
				task = new ImportDataFromAdmCallable(futAdm.get());
				completionServiceData.submit(task);
			}
			executorData.shutdown();
			executorData.awaitTermination(120, TimeUnit.SECONDS);
			log.info ("fin du traitement multiThreads...");

Merci