Discussion :

[nonoRedDevils]

Bonjour!
Tout d'abord je vous explique le contexte de mon projet.
J'ai développé un programme qui collecte des données de differentes, et doit les transmettre à broker (activemq pour ne pas le nommer). Afin d'éviter les goulots d'etranglements, j'ai mis en place entre ces deux composants, une fifo (LinkedBlockingQueue).
Pour l'ecrite et la lecture de message dans cette fifo, j'utilise des threads, et un executor pour regir tout ce beau monde.
Ma grande question est tout bếte : n'aurais je pas un peu abusé des threads, suis je vraiment obligé d'utiliser autant de processus si c'est juste pour qu'ils attendent (utilisation d'un while(true) {sleep()} )
Merci bien !
Le code de mon gestionnaire de fifo :

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
public class FifoManager {
	private Executor executor = null;
	private int nbthread = 10; 
	private LinkedBlockingQueue<String> listeMessage = null;
	private int nbInitialFifoSize = 10000;
	private List<HandlerWriter> handlerWriterList;
	private List<HandlerReader> handlerReaderList;
	public FifoManager() {
		handlerWriterList = new ArrayList<HandlerWriter>();
		handlerReaderList = new ArrayList<HandlerReader>();
		executor = Executors.newFixedThreadPool(nbthread);
		listeMessage = new LinkedBlockingQueue<String>((int) nbInitialFifoSize);
		handlerWriterList.add(new HandlerWriter(listeMessage,"writer"));
		Thread threadWriter = new Thread(handlerWriterList.get(0));
		for(int indexThread = 0; indexThread < nbthread; indexThread ++) {
			HandlerReader handlerReader = new HandlerReader(listeMessage, "reader-"+indexThread);
			handlerReaderList.add(handlerReader);
			executor.execute(handlerReader);
		}
		threadWriter.start();
	}
 
	public void addReaderListener(FifoListenerReader fifoListener) {
		for(HandlerReader handlerReader : handlerReaderList) {
			handlerReader.addListener(fifoListener);
		}
	}
 
	public void pushMessage(String msg){
		for(HandlerWriter handlerWriter : handlerWriterList) {
			handlerWriter.pushMessage(msg);
		}
	}
}

Les threads d'ecriture :

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
public class HandlerWriter implements Runnable {
	private LinkedBlockingQueue<String> listeMessage;
	private String name;
	public HandlerWriter(LinkedBlockingQueue<String> listeMsg, String name) {
		listeMessage = listeMsg ;
		this.name = name;
	}
	@Override
	public void run() {
		while(true){
			try{
				Thread.sleep(100);
			}catch(InterrruptedException e){}			
		}
	}
 
	public void pushMessage(String message) {
		try {
			listeMessage.put(message);
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
 
}

Les threads de lecture :

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
public class HandlerReader  implements Runnable {
	private LinkedBlockingQueue<String> listeMessage;
	private List<FifoListenerReader> listListener;
	private String name;
	public HandlerReader( LinkedBlockingQueue<String> listeMsg, String name) {
		listeMessage = listeMsg;
		listListener = new ArrayList<FifoListenerReader>();
		this.name = name;
	}
 
	@Override
	public void run() {
		while(true) {
			try {
				String msg= listeMessage.take();
				notifyListeners(msg);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}
	}
 
	public void addListener(FifoListenerReader fifoListener) {
		if(fifoListener != null) {
			listListener.add(fifoListener);
		}
	}
 
	public void notifyListeners(String message) {
		for(FifoListenerReader fifoListener : listListener) {
			fifoListener.messageAvalaible(name +":"+message);
		}
	}
}

[nonoRedDevils]

hum hum....
79 vues en 2jours, 0 réponse.
Aurais je écrit de telles enormités que ca ne mérite meme pas une réponse?

[thelvin]

hum hum....
79 vues en 2jours, 0 réponse.
Aurais je écrit de telles enormités que ca ne mérite meme pas une réponse?

Euh, nan mais pas forcément mais comment dire... Ça sert à quoi ? C'est quoi le principe de toutes ces classes, là ?

T'as pas une explication un peu plus naturelle que 8 pages de code Java multithreadé ? Des dessins ? Un exemple d'utilisation ?... C'est quoi le rapport avec des goulots d'étranglement ?... Ce genre de choses.

[nonoRedDevils]

Euh, nan mais pas forcément mais comment dire... Ça sert à quoi ? C'est quoi le principe de toutes ces classes, là ?

T'as pas une explication un peu plus naturelle que 8 pages de code Java multithreadé ? Des dessins ? Un exemple d'utilisation ?... C'est quoi le rapport avec des goulots d'étranglement ?... Ce genre de choses.

ouep ouep...pas faux
Le but de ce logiciel est de collecter des données venant de capteurs, et de les envoyer un Mom.
Les sources peuvent balancer en parallele et à gros débit, pour eviter les pertes de données (de ce que j'ai observé l'envoie sur un broker comme activemq met en moyenne 3milli secondes), on utilise une fifo intermediare pour absorber la charge.
Toute la question est de savoir comment empiler et depiler des messages dans cette fifo.
J'avais donc utilisé une linkedbloquingqueue, et donc un pool de thread.
Ma plus grande question est de savoir si c'est bien intelligent d'avoir des threads qui tournent en rond, et qui n'attendent juste qu'un message arrive de mes sources de données pour les empiler dans la fifo.
Merci
Désolé c'etait un peu brute de fondrerie comme premiere approche...j'avoue...

[professeur shadoko]

c'est marrant j'ai un problème analogue .... mais ma première solution a été de développer un système alternatif à la queue de message (j'utilise JGroups sur UDP).
je n'en suis pas encore à avoir un structure de bufferisation des messages en attente.
Question: pourquoi avoir autant de Threads? chacun de tes canaux d'évènement ne peut pas déposer son évènement dans une structure d'attente et un seul Thread les consomme et les envoie sur la queue de message (de toute façon il faudra limiter ce buffer et prévoir des "étouffements".