Apache
/
activemq
mirror of https://github.com/apache/activemq.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

https://issues.apache.org/jira/browse/AMQ-5558 change on producer / consumer - One session per thread 

I'm backporting a change I have made into Artemis for this. Each thread should have its own JMS Session
This commit is contained in:
Clebert Suconic 2015-06-26 10:19:21 -04:00
parent b64b8ba27e
commit c5579ff73d
2 changed files with 12 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
activemq-console/src/main/java/org/apache/activemq/console/command/ConsumerCommand.java
View File

@ -60,17 +60,16 @@ public class ConsumerCommand extends AbstractCommand {
} }
conn.start(); conn.start();
Session sess;
if (transacted) {
sess = conn.createSession(true, Session.SESSION_TRANSACTED);
} else {
sess = conn.createSession(false, ackMode);
}
CountDownLatch active = new CountDownLatch(parallelThreads); CountDownLatch active = new CountDownLatch(parallelThreads);
for (int i = 1; i <= parallelThreads; i++) { for (int i = 1; i <= parallelThreads; i++) {
Session sess;
if (transacted) {
sess = conn.createSession(true, Session.SESSION_TRANSACTED);
} else {
sess = conn.createSession(false, ackMode);
}
ConsumerThread consumer = new ConsumerThread(sess, ActiveMQDestination.createDestination(destination, ActiveMQDestination.QUEUE_TYPE)); ConsumerThread consumer = new ConsumerThread(sess, ActiveMQDestination.createDestination(destination, ActiveMQDestination.QUEUE_TYPE));
consumer.setName("consumer-" + i); consumer.setName("consumer-" + i);
consumer.setDurable(durable); consumer.setDurable(durable);

13
activemq-console/src/main/java/org/apache/activemq/console/command/ProducerCommand.java
View File

@ -60,16 +60,15 @@ public class ProducerCommand extends AbstractCommand {
conn = factory.createConnection(user, password); conn = factory.createConnection(user, password);
conn.start(); conn.start();
Session sess;
if (transactionBatchSize != 0) {
sess = conn.createSession(true, Session.SESSION_TRANSACTED);
} else {
sess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
}
CountDownLatch active = new CountDownLatch(parallelThreads); CountDownLatch active = new CountDownLatch(parallelThreads);
for (int i = 1; i <= parallelThreads; i++) { for (int i = 1; i <= parallelThreads; i++) {
Session sess;
if (transactionBatchSize != 0) {
sess = conn.createSession(true, Session.SESSION_TRANSACTED);
} else {
sess = conn.createSession(false, Session.AUTO_ACKNOWLEDGE);
}
ProducerThread producer = new ProducerThread(sess, ActiveMQDestination.createDestination(destination, ActiveMQDestination.QUEUE_TYPE)); ProducerThread producer = new ProducerThread(sess, ActiveMQDestination.createDestination(destination, ActiveMQDestination.QUEUE_TYPE));
producer.setName("producer-" + i); producer.setName("producer-" + i);
producer.setMessageCount(messageCount); producer.setMessageCount(messageCount);