Réception JMS avec du multi-threading
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:
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:
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) |
