ExecutorService Producteur/ consomateur

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:

---

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:

---

1 2 3 4

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

---

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:

---

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)); }

---

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:

---

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