Discussion :

[stoukou]

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 :

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

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

[Mobius]

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 :

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

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 :

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

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 WorkQueue.java :

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
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

[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 :

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

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

un thread:

Code :

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

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 :

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

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

[michon]

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

Je suis fan de cette précision dans les termes

[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

[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.

[Mobius]

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_

[stoukou]

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

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 :

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

[Mobius]

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

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 ?

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 :

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
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.

[tchize_]

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 :

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

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

[LGnord]

Tu as aussi les outils type Ateji PX. Tu écris alors quelques choses comme:

Code :

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

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.