Discussion :

[vonemya]

Bonjour,

J'essayer de mettre en place un client multi-threadé et qui consomme des jms.
J'obtiens une erreur "bizarre".
Je sais que dasn un contexte multi-threadé, il faut disposer d'une session par thread car les sessions ne sont pas thread-safe selon les spécification JMS.
Sauf que quand un thread essaye de créer sa propre session, j'obtiens une exception indiquant que la javax.jms.connection est dans un état incohérent alors que sont instanciation s'est bien passée

Avez-vous déjà rencontré ce problème? Avez-vous une idée?

merci beaucoup

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
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
 
 
public class ReceiveJmsDemoMultiThreaded {
 
	public static void main(String[] args) {
		Context context = null;
		ConnectionFactory factory = null;
		Connection connection = null;
		Destination destination = null;
 
		try {
			context = getInitialContext();
			factory = (QueueConnectionFactory) context.lookup("JQCF");
			destination = (Destination) context.lookup("sampleQueue");
			connection = factory.createConnection();
 
			final ExecutorService executor = Executors.newCachedThreadPool();
			executor.submit(new MessageWorker(connection, destination) );
 
			executor.submit(new MessageWorker(connection, destination) );
 
			executor.submit(new MessageWorker(connection, destination) );
 
			executor.submit(new MessageWorker(connection, destination) );
 
 
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			if (context != null) {
				try {
					context.close();
				} catch (NamingException e) {
					e.printStackTrace();
				}
			}
 
			if (connection != null) {
				try {
					connection.close();
				} catch (JMSException e) {
					e.printStackTrace();
				}
			}
		}
	}
 
	private static InitialContext getInitialContext() throws NamingException {
		Properties prop = new Properties();
		prop.put("java.naming.provider.url", "rmi://localhost:1099");
		// prop.put("java.naming.factory.initial",
		// "org.ow2.carol.jndi.spi.MultiOrbInitialContextFactory");
		prop.put("java.naming.factory.initial",
				"org.objectweb.carol.jndi.spi.MultiOrbInitialContextFactory");
		return new InitialContext(prop);
	}
 
}
 
class MessageWorker extends Thread {
	Connection connection = null;
	Destination dest = null;
	Session session = null;
	Destination destination = null;
 
	public MessageWorker(Connection connection, Destination dest) {
		this.connection = connection;
		this.destination = dest;
	}
 
	/**
         * @see java.lang.Thread#run()
         */
	@Override
	public void run() {
		try {
			MessageConsumer receiver = null;
			System.out.println("Starting Thread "+currentThread().getName());
			while (true) {
				try {
					System.out.println("Waiting for next msg "+currentThread().getName());
					session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
					receiver = session.createConsumer(destination);
					connection.start();
					Message msg = receiver.receive();
					if (msg instanceof Message && msg != null) {
						System.out.println("STARTING consuming "+msg.toString()+" by thread  "+currentThread().getName() );
						Thread.sleep(2000);//some work here
						System.out.println("ENDING consuming "+msg.toString()+" by thread  "+currentThread().getName() );
					}
				} catch (JMSException e) {
					e.printStackTrace();
				}
			}
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
 
		}
	}
}

et voici l'erreur que j'obtiens:

Code :

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

1
2
3
4
5
6
7
8
9
10
11
12
 
 
javax.jms.IllegalStateException: Forbidden call on a closed connection.
	at org.objectweb.joram.client.jms.Connection.checkClosed(Connection.java:404)
	at org.objectweb.joram.client.jms.Connection.createSession(Connection.java:530)
	at MessageWorker.run(ReceiveJmsDemoMultiThreaded.java:96)
	at java.util.concurrent.Executors$RunnableAdapter.call(Unknown Source)
	at java.util.concurrent.FutureTask$Sync.innerRun(Unknown Source)
	at java.util.concurrent.FutureTask.run(Unknown Source)
	at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(Unknown Source)
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(Unknown Source)
	at java.lang.Thread.run(Unknown Source)

[tchize_]

dans ton finally tu fait un close de la connection. Mais comme à aucun moment tu n'attends que ton executor service n'aie fini le travail soumis, tu ferme ta connection avant que le autre threads aien pu faire quoi que ce soit.

tu peux mettre un executor.awaitTermination(timeout,unit) pour laisser le temps aux threads de finir.

Ou, alternativement, tu appelle la méthode get() sur chacun des Future<> retournés par submit().

[vonemya]

Effectivement, il fallait attendre la fin de l'executor avant de faire connection.close() dans la close finally.

Il manquait donc cette instruction:

Code :

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

1
2
 
executor.awaitTermination(1, TimeUnit.DAYS)

Merci pour ton aide

[manblaizo]

Je ne suis pas sûr que le problème soit résolu, car l'appel

Code :

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

executor.awaitTermination(1, TimeUnit.DAYS)

va attendre un jour et sortir en retournant false, dû à un timeout. Et comme ton MessageWorker est dans une boucle infinie, l'exception IllegalStateException devrait réapparaître à un moment ou un autre.

En plus, en principe, executor.awaitTermination() est appelé après un shutdown de l'ExecutorService (après appel à executor.shutdown()), parce qu'il s'agit en fait d'attendre que les threads lancés terminent leurs tâches respectives avant de stopper l'ExecutorService.

Il faudrait donc trouver un moyen pour faire durer tes objets Connection, Destination et ExecutorService le temps nécessaire à l'exécution des messages, peut-être avec une boucle infinie dans la méthode main(), mais dans ce cas, il faudra bien alors réfléchir à une procédure pour arrêter l'application ...
Tes MessageWorker devraient aussi être implémentés de manière à ce qu'ils soient interruptibles, car ce sont des threads non daemon, ce qui empêcherait l'instance de la JVM de jamais se terminer ...

Bref, à mon avis, ce n'est pas aussi simple qu'un executor.awaitTermination(). Il faudrait analyser un peu plus la problématique, en fonction de ton vrai besoin.

[tchize_]

Comme je l'ai dit, suffit pour chaque Future retourné de demander la valeur de retour. Ainsi on est sur que chaque Future et terminé et là on peux tout cloturer.

Si t'es sur que les tâches ne peuvent par partir en sucette (attente infinie) alors utiliser l'executor server avec son shutdow / awaitTerminaison est tout aussi correct.