[Débutant] Threads boucle for

Version imprimable

Bonjour,
voilà mon problème j'ai un ListPersonnes qui contient une liste de personne loadé depuis un fichier xml avec Jaxb
donc moi je voulais faire un certain traitement pour chaque ligne (personne)

mon objet ListPersonnes contient 50000 ligne (personne) le traitement met 7 heure donc j'ai pensé à un processe mutithread
et mon idée c'est de traiter les ligne par lots de 10 par exemple
du j'ai pensé à boucle qui ressemble à ça mais le soucis comme c'est je dois l'astuce qui va me permettre de permettre de lancer par lot

Merci de votre aide
Code:

1 2 3 4 5 6 7 8 9 10 11 12 13 for (int i = 0; i < personnes.getpersonne().size(); i++) { personne = personnes.getpersonne().get(i); Thread r = new Thread() { @Override public void run() { PersonneUtil.processpersonne(connexion, getRemotePers(connexion), errorParameter, personne); } }; r.start(); }

Pour répondre directement a ta question tu peux faire quelque chose comme ca (attention le code est erroné, les condition de la boucle ne sont pas juste et les index pour récupérer les sous listes sont également faux) :

Code:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
		for (int i = 0; i < personnes.getpersonne().size() / 10 ; i++) {
			final List<Personne> lstPersonnes = personnes.getpersonne().subList(10*i, 11*i);
			Thread r = new Thread() {
				@Override
				public void run() {
					for (Personne personne : lstPersonnes) {
						PersonneUtil.processpersonne(connexion,
								getRemotePers(connexion), errorParameter, personne);
					}
				}
 
			};
			r.start();
		}

Une autre facon de faire est de limiter le nombre de thread (ca ne sert a rien d'avoir plus de thread que de processeur) et que chaque thread récupère du job quand il a terminé son travail.
Voici un exemple inspiré d'un article d'IBM

Code:

1
2
3
4
5
6
7
8
9
10
		WorkQueue queue = new WorkQueue(4);
		for (int i = 0; i < personnes.getpersonne().size(); i++) {
			personne = personnes.getpersonne().get(1);
			queue.execute(new Runnable() {
				public void run() {
					PersonneUtil.processpersonne(connexion,
							getRemotePers(connexion), errorParameter, personne);
				}
			});
		}

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
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
import java.util.LinkedList;
 
public class WorkQueue
{
    private final int nThreads;
    private final PoolWorker[] threads;
    private final LinkedList queue;
 
    public WorkQueue(int nThreads)
    {
        this.nThreads = nThreads;
        queue = new LinkedList();
        threads = new PoolWorker[nThreads];
 
        for (int i=0; i<nThreads; i++) {
            threads[i] = new PoolWorker();
            threads[i].start();
        }
    }
 
    public void execute(Runnable r) {
        synchronized(queue) {
            queue.addLast(r);
            queue.notify();
        }
    }
 
    private class PoolWorker extends Thread {
        public void run() {
            Runnable r;
 
            while (true) {
                synchronized(queue) {
                    while (queue.isEmpty()) {
                        try
                        {
                            queue.wait();
                        }
                        catch (InterruptedException ignored)
                        {
                        }
                    }
 
                    r = (Runnable) queue.removeFirst();
                }
 
                // If we don't catch RuntimeException, 
                // the pool could leak threads
                try {
                    r.run();
                }
                catch (RuntimeException e) {
                    // You might want to log something here
                }
            }
        }
    }
}

Vu que j'ai écrit ca rapidement, c'est loin d'être parfait

21/01/2011, 11h47
tchize_
pour faire les traitements en //, vous démarrez N+1 thread

Les N premiers threads vont lire dans une SynchronousQueue les personnes à traiter, et le dernier thread va y écrire au fur et à mesure le contenu de la list.
Grosso modo:

Code:

SynchrounousQueue<Personne> synchronousQueue=new SynchrounousQueue<Personne>();

un thread:
Code:

1 2 3 4 for (Personne personne: listePersonne) synchronousQueue.put(personne); for (int i=0;i<nbreThreads) synchronousQueue.put(UN_MARQUEUR_DE_FIN); // ben oui, un marqueur par thread
les autres threads
Code:

1 2 3 4 Personne personne = null; while ((personne=synchrounousQueue.take()!=UN_MARQUEUR_DE_FIN) traiter(personne);
et tout ce bazard finira a peu près son bordel en meme temps
21/01/2011, 11h59
michon

Citation:

Envoyé par tchize_

et tout ce bazard finira a peu près son bordel en meme temps

Je suis fan de cette précision dans les termes :mrgreen:
21/01/2011, 12h04
stoukou

hello merci à vous deux pour vos réponses
j'ai commencer à tester la solution du QueuWorker de Mobius (je viens de voir ta réponse tchize_ )

ce j'arrive pas à comprende c'est que le WorkQueue traite seulement le nombre de ligne qu'on passe en parametre ??? bizarre alors qu'en regardant le code j'ai l'impression il doit mettre ça dans la linked liste et traite un par un

sinon tchize_ est ce que ta sollution lance tout les thrad en même temps? si c'est le cas ça risque de m'ouvrir un nombre énorme de connexion vers ma base et ça rique de la faire tomber non?

merci encore une fois de votre aide
21/01/2011, 12h18
tchize_

ben c'est vous qui choisissez combien de threads vous voulez, je vous ai juste mis le code à mettre dans les thread

A priori, rien ne sert de faire plus de thread qu'il n'y a d'unité de calcul dans la machine, sauf si le but est de palier à des lenteur réseau ou autre. Mais bon, a moin de lancer 500 thread en // vosu risquez pas de causer de problème à la DB.
21/01/2011, 14h08
Mobius

Citation:

Envoyé par stoukou

ce j'arrive pas à comprende c'est que le WorkQueue traite seulement le nombre de ligne qu'on passe en parametre ??? bizarre alors qu'en regardant le code j'ai l'impression il doit mettre ça dans la linked liste et traite un par un

Le WorkingQueue, traite toutes les personnes.
Dans l'exemple, seulement 4 threads seront créés. Les 4 Threads traiteront toutes les personnes au fur et à mesure. Cela permet de réaliser 4 traitements en parallèle.
Il s'agit d'un solution très similaire à celle de tchize_

Citation:

Envoyé par Mobius

Le WorkingQueue, traite toutes les personnes.
Dans l'exemple, seulement 4 threads seront créés. Les 4 Threads traiteront toutes les personnes au fur et à mesure. Cela permet de réaliser 4 traitements en parallèle.
Il s'agit d'un solution très similaire à celle de tchize_

oui justement c'est que j'ai compris du code mais mon soucis c'est que c'est pas le cas en exécutant :oops:

bon je suis entrain de tester une autre piste c'est le ThreadpoolExecutor qu'est ce que vous en pensee de ce code et la question qui tue comment je lance mes thread maintenant pour qu'il me lance 5 thread en paralleles
Code:

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 BlockingQueue<Runnable> bqueue = new SynchronousQueue<Runnable>(); ThreadPoolExecutor execut = new ThreadPoolExecutor(5, 5, 10, TimeUnit.SECONDS, bqueue); for (int i = 0; i < personnes.getpersonne().size(); i++) { personne = personnes.getpersonne().get(i); try { bqueue.put(new Runnable() { @Override public void run() { FirepersonneUtil.processPersonne(ConnectServer.getDefault(), getRemotePers(ConnectServer.getDefault()), errorParameter, personne); } }); } catch (InterruptedException e) { Log.error("FirepersonneUtil", e); }

Citation:

Envoyé par stoukou

oui justement c'est que j'ai compris du code mais mon soucis c'est que c'est pas le cas en exécutant :oops:

Il est possible que ton programme s'arrete avant que les Threads ait finis tout le travail.
Il faut bien voir que ton thread principal, après avoir initialisé ton QueueWorker, il est finit. Le programme va donc s'arrêter (et dommage pour les autres Thread qui bossent). Il faut donc trouver une méthode pour attendre la fin du travail de tous les threads.
Est ce que c'est ca ton problème ou c'est encore autre chose ?
Citation:
Envoyé par stoukou
bon je suis entrain de tester une autre piste c'est le ThreadpoolExecutor qu'est ce que vous en pensee de ce code et la question qui tue comment je lance mes thread maintenant pour qu'il me lance 5 thread en paralleles
Code:

1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 BlockingQueue<Runnable> bqueue = new SynchronousQueue<Runnable>(); ThreadPoolExecutor execut = new ThreadPoolExecutor(5, 5, 10, TimeUnit.SECONDS, bqueue); for (int i = 0; i < personnes.getpersonne().size(); i++) { personne = personnes.getpersonne().get(i); try { bqueue.put(new Runnable() { @Override public void run() { FirepersonneUtil.processPersonne(ConnectServer.getDefault(), getRemotePers(ConnectServer.getDefault()), errorParameter, personne); } }); } catch (InterruptedException e) { Log.error("FirepersonneUtil", e); }
C'est une bonne idée d'utiliser un ThreadPoolExecutor.
Cependant tu risque d'avoir le même genre de problème.

Dans ton code, il y a une erreur. Il ne faut pas rajouter la tache directement dans la queue mais il faut utiliser la méthode execute de ton ThreadPoolExecutor.

Citation:

Envoyé par Mobius

Il est possible que ton programme s'arrete avant que les Threads ait finis tout le travail.
Il faut bien voir que ton thread principal, après avoir initialisé ton QueueWorker, il est finit. Le programme va donc s'arrêter (et dommage pour les autres Thread qui bossent). Il faut donc trouver une méthode pour attendre la fin du travail de tous les threads.

Alors, a mois d'appeler System.exit ou que les threads faisant le traitement soient en daemon, le programme ne se terminera pas.

Ensuite, pour attendre la fin de tous les thread, il n'y a pas à se casser la nenettes

Grosso merdo
Code:

1 2 3 4 5 6 7 8 9 10 11 Thread remplissage = ....; Thread[] traitements = ......; remplissage.start(); for (Thread t: traitements) t.start(); // tout le traitement est en route maintenant, on attends la fin remplissage.join(); for (Thread t: traitements) t.join(); // tout est fini

24/01/2011, 09h47
LGnord
Tu as aussi les outils type Ateji PX. Tu écris alors quelques choses comme:
Code:

1 2 3 4 for || (int i = 0; i < personnes.getpersonne().size(); i++) { PersonneUtil.processpersonne(personne); }
La gestion des threads est à la chage de l'outils.