Discussion :

[Bktero]

Bonjour,

Je suis en train de réfléchir à un moyen de synchroniser des threads. Ces threads traitent des enregistrements (lus dans des fichiers) avant de les insérer dans une base de données. Pour le moment, les threads ignorent ce que traitent les autres threads à ce moment là. On peut se retrouver avec des lignes en doublons dans la table (certains champs de l'enregistrement constituant une clé de dédoublonnage).

Une première idée que j'ai mûri avec un collègue est d'utiliser une HashTable, qui prend en clé la clé de dédoublonnage des enregistrements et en valeur une instance d'une classe maison. Cette classe, appelé Enregistrement (rien à voir avec l'enregistrement traité, le nom n'est pas choisi) contient un objet quelconque qui sert de mutex ainsi que d'autres informations utiles.

Chaque thread va lire la HashTable pour déterminer si un enregistrement similaire au sien est en cours de traitement.
- Si non, il écrit dans la hashtable et prend le mutex sur l'objet qu'il vient d'insérer. Si le relache quand il a finit.
- Si oui, on demande le mutex sur l'objet existant et on attend de l'obtenir pour traiter.

Chaque thread fait des pré-traitements sur son enregistrement, puis appelle la fonction suivante (code test et non code définitif, d'où des sleep à la place de vrais traitements) :

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
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
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
private static HashTable<Integer, Enregistrement> synchronizer  = new HashTable<Integer, Enregistrement>();
          public boolean utiliserSynchronizer() {
 
	  boolean estPresent = false;
 
	  // Creation de l'enregistrement a mettre la Synchronizer
	  Enregistrement monRec = new Enregistrement(threadMessage);
 
	// Creation de l'enregistrement qu'on recuperera du Synchronizer
	  Enregistrement existant;
 
	  // Verrouillage du Synchronizer ( http://rom.developpez.com/java-synchronisation/#LII-A )
	  synchronized(synchronizer) {
 
		  if(true==synchronizer.containsKey(threadId)) {
			  estPresent = true;
		  }
		  else {
			  synchronizer.put(threadId, monRec);
			  estPresent = false;
		  }
	  }
	  /* On libere au plus vite la HashTable. Ainsi, d'autres threads peuvent y accéder.
	   * Dans la suite, on fait l'action necessaire en fonction de la variable "estPresent"
	   */
	  if(false==estPresent) {
		  synchronized(monRec.getVerrou()) {
			  // On simule le traitement d'encodage du record
			  try {
				  System.out.println("On commence a encoder (" + threadMessage + ")");
				  sleep(10000);
				  System.out.println("On termine d'encoder (" + threadMessage + ")");
			  }
			  catch(Exception ex){}
		  }		  
	  }
	  else {
		  // Recuperation de cet enregistrement existant
		  existant = synchronizer.get(threadId);
		  System.out.println( "\tJ'ai lu : " + existant.toString() + "(" + threadMessage + ")");	
 
		  // Comparer la date
		  //if( monRec.getDate().after(existant.getDate())==false ) {
		  if(false) {
			  //System.out.println("On rejette l'enregistrement  (" + threadMessage + ")");
		  }
		  else { 
			  // On se met en attente de l'enregistrement
			  synchronized(existant.getCompteur()) {
				  existant.setCompteur(existant.getCompteur() + 1);
			  }	  	  
 
			  // On attend notre tour...
			  System.out.println("On attend notre tour  (" + threadMessage + ")");
			  synchronized(existant.getVerrou()) {
				  System.out.println("On a obtenu le verrou  (" + threadMessage + ")");
				  // On n'est plus en attente donc on decremente le verrou
				  existant.setCompteur(existant.getCompteur() - 1);
 
				  // On remplace avec nos donnees
				  existant.setDebugTxt( existant.getDebugTxt() + "_" + monRec.getDebugTxt() );
 
				  // On simule le traitement d'encode du record
				  try {
					  System.out.println("On commence a encoder (" + threadMessage + ")");
					  sleep(10000);
					  System.out.println("On termine d'encoder (" + threadMessage + ")");
				  }
				  catch(Exception ex){}					  					  					  					  
			  } // section critique sur l'enregistrement
		  }  
	  }//esPresent ?
	  return estPresent;
  }

Ca fonctionne bien à un gros détail près : l'appel à synchronized(monRec.getVerrou()) bloque complètement l'accès à la hashtable.

Conséquence : 2 threads ne peut pas être en cours d'encodage en même temps. C'est pas terrible

Auriez-vous une suggestion pour contourner ce problème ? Merci d'avance !