EzDevInfo.com

Real world use of JMS/message queues? [closed]

I was just reading abit about JMS and Apache ActiveMQ. And was wondering what real world use have people here used JMS or similar message queue technologies for ?

What design decisions would favour Scala's Actors instead of JMS?

What are the differences using Scala Actors instead of JMS?

For example from a performance and scalability perspective, what does the Scala Actor model add compared to JMS? In which cases does it make more sense to use Actors rather than JMS, i.e. what problems does Actors address that JMS cannot cover?

Why are SpyJMSExceptions still thrown after recycling the client JBoss connection to remote queues?

My application below communicates as a client on a JBoss 7.2.0 system to a receiver JNDI/JMS on a JBoss 4.2.1 system. It creates a Send Queue and a Receive Queue. We have been running just fine for 2 months straight with this configuration; no changes were made to either side. The local client app has the 4.2.1 jbossall-client.jar and jnp-client.jars installed.

After normal activity, we start receiving a org.jboss.mq.SpyJMSException: Exiting on IOE; - nested throwable: (java.io.EOFException) at org.jboss.mq.SpyJMSException.getAsJMSException(SpyJMSException.java:72) exception.

We restarted JBoss 7.2.0 without changing anything, and when we are establishing the receive queue, we are now receiving a org.jboss.mq.SpyJMSException: Cannot subscribe to this Destination: ; {...} Caused by: java.io.EOFException exception thrown from our code at QueueReceiver receiver = session.createReceiver(queue);. We also start throwing this same exception after the app has been running fine for days, but without connectivity over a multi-day period of time.

We had the 4.2.1 system restarted to see if that was the problem, but that fixed nothing. In fact, we can replicate this failure scenario by having both systems connect as normal, then recycle the 4.2.1 system. The errors start throwing once the 4.2.1 system is down, and the 7.2.0 system continues to fail to re-establish the connections once the 4.2.1 system is fully established (even when it SHOULD be capable of doing so).

Stopping then starting the app in JBoss does not fix this. Restarting JBoss has a 20% chance of fixing this (100% chance in the case of the forced failure scenario cited above). Undeploying, then redeploying the application usually fixes this.

What might be causing this?

This same war file works fine on our test system, which has an identical JBoss set-up. Communication with the target JBoss system via a test app from the command prompt using the same code works fine.

I suspect there is an issue with JBoss 7.2.0 itself, or is this possibly a timeout issue? How do I check or extend the timeout length; is that something doable from the client side? Even if it was timeout, I make the call to the stop() method before the rest of start() re-connects, and I still get the exception; in that scenario, it wouldn't be a timeout issue as a timeout would have reset.

Context Values:

connectionFactoryName=ConnectionFactory
contextFactoryName=org.jnp.interfaces.NamingContextFactory
packagePrefixes=org.jboss.naming:org.jnp.interfaces
providerUrl=MYSERVER:1099

Java Code:

private ContextContainer contextContainer = null;
private QueueConnection connection = null;
private QueueReceiver receiver = null;
private Queue sendQueue = null;
private Queue receiveQueue = null;
private String sendQueueName = null;
private String receiveQueueName = null;
private MessageHandler messageHandler = null;

protected synchronized void start() throws Exception {

    // NOTE: This position also has delay code (for pending multiple 
    // consecutive recycling requests), with an external method to 
    // reset the delay. It was removed for code clarity and has no 
    // effect on the issue being discussed.

    // Clear prior Connection
    stop();

    logger.info("Regenerating JMS for : " + this.getClass().getName());

    Context context = this.contextContainer.getContext();

    logger.info("Looking up factory : " + contextContainer.getConnectionFactoryName());

    QueueConnectionFactory connectionFactory = (QueueConnectionFactory)context.lookup(contextContainer.getConnectionFactoryName()); 

    // ESTABLISH SEND MESSAGE QUEUE
    logger.info("Looking up send queue : " + sendQueueName);

    sendQueue = (Queue)context.lookup(sendQueueName); 

    logger.info("Send Queue string : " + sendQueue);
    logger.info("Send Queue name   : " + sendQueue.getQueueName());
    logger.info("Creating Queue Connection");

    connection = connectionFactory.createQueueConnection();

    logger.info("Setting Exception Listener");

    connection.setExceptionListener(new ExceptionListener() {

        public void onException(JMSException ex) {

            logger.error("JMS Exception received on QueueConnection", ex);

            start();
        }
    });

    // ESTABLISH RECEIVE MESSAGE QUEUE
    logger.info("Looking up receive queue : " + receiveQueueName);

    receiveQueue = (Queue)context.lookup(receiveQueueName); 

    logger.info("Receive Queue string : " + receiveQueue);
    logger.info("Receive Queue name   : " + receiveQueue.getQueueName());
    logger.info("Creating JMS Session for Receiving");

    QueueSession session = connection.createQueueSession(false, Session.CLIENT_ACKNOWLEDGE);

    logger.info("Created Session " + session);      
    logger.info("Creating JMS Receiver for Queue \"" + receiveQueue.getQueueName() + "\"");

    // THIS IS THE LINE WHERE THE EXCEPTION IS THROWN!!!
    receiver = session.createReceiver(receiveQueue);

    logger.info("Setting Message Listener");

    receiver.setMessageListener(new MessageListener() {

        public void onMessage(Message message) {

            try {

                if (message instanceof ObjectMessage) {

                    Object obj = ((ObjectMessage) message).getObject();

                    //  UNRELATED METHOD FOR HANDLING THE MESSAGE
                    handleMessage(obj);

                } else {

                    throw new Exception("Received message of unexpected type " + message.getJMSType() + ", was expecting ObjectMessage");
                }

            } catch (Exception ex) {

                logger.error("Error processing incoming message.", ex);

            } finally {

                try {

                    message.acknowledge();
                    logger.info("Acknowledged message.");

                } catch (Exception ex) {

                    logger.error("Unable to acknowledge message.", ex);
                }
            }
        }
    });

    logger.info("Starting Queue Connection");

    connection.start();

    logger.info("Started Queue Connection");
}   

/**
 * Extinguish the existing connection.
 */
public synchronized void stop() {

    if (receiver != null) {

        try {

            logger.info("Nullifying Receiver Listener");
            receiver.setMessageListener(null);
            logger.info("Nullified Receiver Listener");

        } catch(Exception ex) {

            logger.warn("Exception nullifying Receiver Listener", ex);
        }

        try {

            logger.info("Closing Receiver");
            receiver.close();
            logger.info("Closed Receiver");

        } catch(Exception ex) {

            logger.warn("Exception closing Receiver", ex);

        } finally {

            receiver = null;
        }           
    }

    if (connection != null) {

        try {

            logger.info("Nullifying Exception Listener");
            connection.setExceptionListener(null);
            logger.info("Nullified Exception Listener");

        } catch (Exception ex) {

            logger.warn("Exception nullifying Exception Listener", ex);
        }

        try {

            logger.info("Stopping Queue Connection");
            connection.stop();
            logger.info("Stopped Queue Connection");

        } catch (Exception ex) {

            logger.warn("Exception stopping Queue Connection", ex);
        }

        try {

            logger.info("Closing Queue Connection");
            connection.close();
            logger.info("Closed Queue Connection");

        } catch (Exception ex) {

            logger.warn("Exception closing Queue Connection", ex);

        } finally {

            connection = null;
        }
    }
}

Log Output:

Setting Context Factory Class: org.jnp.interfaces.NamingContextFactory
Setting Package Prefixes: org.jboss.naming:org.jnp.interfaces
Setting Provider URL: MYSERVER:1099
Generating JMS for : MYPACKAGE.ConnectionHandler
Looking up factory : ConnectionFactory
Looking up send queue : queue/sendQueue
Send Queue string : QUEUE.sendQueue
Send Queue name   : sendQueue
Creating Queue Connection
Setting Exception Listener
Looking up receive queue : queue/receiveQueue
Receive Queue string : QUEUE.receiveQueue
Receive Queue name   : receiveQueue
Creating JMS Session for Receiving
Created Session SpySession@1903462020[tx=false ack=CLIENT txid=null RUNNING connection=Connection@1963631553[token=ConnectionToken:ID:273725/9966a9625bb094d33a37f72db71b3bb9 rcvstate=STOPPED]]
Creating JMS Receiver for Queue "receiveQueue"
Exception caught during initialization.
org.jboss.mq.SpyJMSException: Cannot subscribe to this Destination: ; - nested throwable: (java.io.EOFException)

Exception trace after normal activity:

JMS Exception received on QueueConnection
org.jboss.mq.SpyJMSException: Exiting on IOE; - nested throwable: (java.io.EOFException)
at org.jboss.mq.SpyJMSException.getAsJMSException(SpyJMSException.java:72)
at org.jboss.mq.Connection.asynchFailure(Connection.java:423)
at org.jboss.mq.il.uil2.UILClientILService.asynchFailure(UILClientILService.java:174)
at org.jboss.mq.il.uil2.SocketManager$ReadTask.handleStop(SocketManager.java:439)
at org.jboss.mq.il.uil2.SocketManager$ReadTask.run(SocketManager.java:371)
at java.lang.Thread.run(Thread.java:744)
Caused by: java.io.EOFException
at java.io.ObjectInputStream$BlockDataInputStream.readByte(ObjectInputStream.java:2766)
at java.io.ObjectInputStream.readByte(ObjectInputStream.java:916)
at org.jboss.mq.il.uil2.SocketManager$ReadTask.run(SocketManager.java:316)
... 1 more

Exception trace after restart:

org.jboss.mq.SpyJMSException: Cannot subscribe to this Destination: ; - nested throwable: (java.io.EOFException)
    at org.jboss.mq.SpyJMSException.getAsJMSException(SpyJMSException.java:72)
    at org.jboss.mq.SpyJMSException.rethrowAsJMSException(SpyJMSException.java:57)
    at org.jboss.mq.Connection.addConsumer(Connection.java:800)
    at org.jboss.mq.SpySession.addConsumer(SpySession.java:947)
    at org.jboss.mq.SpySession.createReceiver(SpySession.java:658)
    at org.jboss.mq.SpyQueueSession.createReceiver(SpyQueueSession.java)
    at org.jboss.mq.SpySession.createReceiver(SpySession.java:647)
    at org.jboss.mq.SpyQueueSession.createReceiver(SpyQueueSession.java)
    at MYPACKAGE.ConnectionHandler.start(ConnectionHandler.java:144)
    at MYPACKAGE.ConnectionHandler.initialize(ConnectionHandler.java:60)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:606)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.invokeCustomInitMethod(AbstractAutowireCapableBeanFactory.java:1414)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.invokeInitMethods(AbstractAutowireCapableBeanFactory.java:1375)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.initializeBean(AbstractAutowireCapableBeanFactory.java:1335)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.doCreateBean(AbstractAutowireCapableBeanFactory.java:473)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory$1.run(AbstractAutowireCapableBeanFactory.java:409)
    at java.security.AccessController.doPrivileged(Native Method)
    at org.springframework.beans.factory.support.AbstractAutowireCapableBeanFactory.createBean(AbstractAutowireCapableBeanFactory.java:380)
    at org.springframework.beans.factory.support.AbstractBeanFactory$1.getObject(AbstractBeanFactory.java:264)
    at org.springframework.beans.factory.support.DefaultSingletonBeanRegistry.getSingleton(DefaultSingletonBeanRegistry.java:222)
    at org.springframework.beans.factory.support.AbstractBeanFactory.doGetBean(AbstractBeanFactory.java:261)
    at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:185)
    at org.springframework.beans.factory.support.AbstractBeanFactory.getBean(AbstractBeanFactory.java:164)
    at org.springframework.beans.factory.support.DefaultListableBeanFactory.preInstantiateSingletons(DefaultListableBeanFactory.java:429)
    at org.springframework.context.support.AbstractApplicationContext.finishBeanFactoryInitialization(AbstractApplicationContext.java:728)
    at org.springframework.context.support.AbstractApplicationContext.refresh(AbstractApplicationContext.java:380)
    at org.springframework.web.context.ContextLoader.createWebApplicationContext(ContextLoader.java:255)
    at org.springframework.web.context.ContextLoader.initWebApplicationContext(ContextLoader.java:199)
    at org.springframework.web.context.ContextLoaderListener.contextInitialized(ContextLoaderListener.java:45)
    at org.apache.catalina.core.StandardContext.contextListenerStart(StandardContext.java:3339)
    at org.apache.catalina.core.StandardContext.start(StandardContext.java:3777)
    at org.jboss.as.web.deployment.WebDeploymentService.doStart(WebDeploymentService.java:156)
    at org.jboss.as.web.deployment.WebDeploymentService.access$000(WebDeploymentService.java:60)
    at org.jboss.as.web.deployment.WebDeploymentService$1.run(WebDeploymentService.java:93)
    at java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:471)
    at java.util.concurrent.FutureTask.run(FutureTask.java:262)
    at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1145)
    at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:615)
    at java.lang.Thread.run(Thread.java:744)
    at org.jboss.threads.JBossThread.run(JBossThread.java:122)
Caused by: java.io.EOFException
    at java.io.ObjectInputStream$BlockDataInputStream.readByte(ObjectInputStream.java:2766)
    at java.io.ObjectInputStream.readByte(ObjectInputStream.java:916)
    at org.jboss.mq.il.uil2.SocketManager$ReadTask.run(SocketManager.java:316)
    at java.lang.Thread.run(Thread.java:744)

Which JMS implementation do you use? [closed]

We are using ActiveMQ 5.2 as our implementation of choice and we picked it a while ago. It performs well enough for our use right now. Since its been a while, I was wondering what other Java Message Service implementations are in use and why? Surely there are more than a few.

Which embedded messaging system -> ActiveMQ or HornetQ

I would appreciate some general pointers and opinions regarding which of the two messaging systems is

• easier to manage
• has less gotchas or magic stuff one needs to know and avoid
• has less overal dependencies
• is simple to work with.

What is the right Maven dependency for javax.jms.* classes?

I need to import javax.jms.* classes. What is the right dependency to include into a Maven project? I'm trying javax.jms:jms:1.1, but no luck (it's pom, not jar).

ps. The only workaround I've found so far is: javax:javaee-api:6.0 (from Maven Central).

How does JMS Receive work internally?

I've been researching various communication technologies/architectures/patterns/implementations (read: buzzwords) including Web Services (WCF, Axis2), ESBs, SOA, and wanted to know more about JMS with regards to messaging.

Conceptually, JMS sounds simple. My take is that it's an intermediate broker which manages messages from publishers and routes them to appropriate subscribers. This is done by queueing messages as they are published, and dequeuing them as they are received.

Question 1: Is my basic understanding of JMS correct?

One of the things that bugs me when reading about technologies is when a certain level of (intentional or unintentional) hand-waving is done about a feature.

Based on my basic understanding, a JMS Provider must be running in order to send or receive messages. My assumption on publishing is that the JMS Provider simply waits until a message is published, then stores it in a queue (memory or database-backed, depending on implementation). However, I am not quite sure how receive works.

Question 2: Does receive (typically) block if no messages are avaiable?

Question 2b: If so, how is blocking achieved? Does the client continuously poll for messages? Does the server simply not respond until a message is published (how does this work without timing out?) Does the provider initiate a call to the recipient?

Question 2c: If not, how does one ensure messages are received in a timely manner, without impacting performance?

The basic description seems to lean towards a single JMS provider to ensure that messages are centrally managed not lost. I can see scaling being an issue.

Question 3: How does JMS scale?

When scaling, I can see there being complexities to ensure that a single message is delivered to all appropriate subscribers, regardless of which physical server receives the message.

Question 3b: How does a JMS implementation ensure reliable delivery in a scaled environment?

Please note that although these questions are related to JMS, they likely apply to any messaging infrastructure. I welcome answers specific to JMS as well as those which are more general or even specific to another technology.

Tool for posting test messages onto a JMS queue? [closed]

Can anyone recommend a tool for quickly posting test messages onto a JMS queue? The tool should allow the user to enter some data, perhaps an XML payload, and then submit it to a queue? I know I could probably knock something up reasonably quickly to do this but I thought I'd ask first before reinventing the wheel. Cheers.

Concurrent Synchronous Request-Reply with JMS/ActiveMQ - Patterns/Libraries?

I have a web-app where when the user submits a request, we send a JMS message to a remote service and then wait for the reply. (There are also async requests, and we have various niceties set up for message replay, etc, so we'd prefer to stick with JMS instead of, say, HTTP)

In How should I implement request response with JMS?, ActiveMQ seems to discourage the idea of either temporary queues per request or temporary consumers with selectors on the JMSCorrelationID, due to the overhead involved in spinning them up.

However, if I use pooled consumers for the replies, how do I dispatch from the reply consumer back to the original requesting thread?

I could certainly write my own thread-safe callback-registration/dispatch, but I hate writing code I suspect has has already been written by someone who knows better than I do.

That ActiveMQ page recommends Lingo, which hasn't been updated since 2006, and Camel Spring Remoting, which has been hellbanned by my team for its many gotcha bugs.

Is there a better solution, in the form of a library implementing this pattern, or in the form of a different pattern for simulating synchronous request-reply over JMS?

Related SO question:

• Is it a good practice to use JMS Temporary Queue for synchronous use?, which suggests that spinning up a consumer with a selector on the JMSCorrelationID is actually low-overhead, which contradicts what the ActiveMQ documentation says. Who's right?

JMS Messaging Performance: Lots of Topics/Queues vs. Extensive Filtering (Message Selectors)

I'm working on a project that is going to make heavy use of JBoss Messaging (JMS). I'm tasked with building an easy to use wrapper around Messaging for other developers and am thinking about using JMS's Message Selectors to provide a filtering technique to keep unnecessary sending of messages to a minimum. I'm curious if anyone has any experience with doing so in regards to performance? My fear is that the JMS provider may get bogged down with the Message Selectors, effectively defeating the whole purpose. It would be much nicer however than creating a long list of Topics/Queues for every message type.

Ultimately I'll undoubtedly end up using some combination of the two, but am concerned with the impact on performance whichever way I lean more towards.

jms producer performance with spring

i created a simple producer consumer simulation based on spring, jms and activemq, i'm trying to reach high performance from both sides, producers and consumers,

Connection settings :

<tx:annotation-driven />
<bean id="transactionManager" class="org.springframework.jms.connection.JmsTransactionManager">
     <property name="connectionFactory"  ref="connectionFactory" />
</bean>

<amq:connectionFactory id="amqConnectionFactory" brokerURL="failover:(tcp://${broker.url}:61616)"  />

<bean id="connectionFactory"
    class="org.springframework.jms.connection.CachingConnectionFactory">
    <property name="targetConnectionFactory" ref="amqConnectionFactory" />
</bean>

<amq:queue id="queue" physicalName="queue" />

<beans:bean id="jsonMessageConverter" class="XXXXX.converter.JsonMessageConverter" />

Consumer settings :

<jms:listener-container concurrency="10"
    acknowledge="auto" prefetch="1" message-converter="jsonMessageConverter" transaction-manager="transactionManager"

    >
    <jms:listener id="queueListener_1" destination="ooIntegrationQueue"
        ref="myMessageListenerAdapter" />
</jms:listener-container>


<beans:bean id="myMessageListenerAdapter"
    class="org.springframework.jms.listener.adapter.MessageListenerAdapter" >
    <beans:property name="delegate" ref="consumer"/>
</beans:bean>


<beans:bean id="consumer" class="XXX.ConsumerImpl"/>

Producer settings :

<beans:bean id="jmsTemplate" class="org.springframework.jms.core.JmsTemplate"
    p:connectionFactory-ref="connectionFactory" p:messageConverter-ref="jsonMessageConverter"
    p:defaultDestination-ref="ooIntegrationQueue" p:sessionTransacted="true" />

starting with the consumer, i managed to consume about 25 messages per second, which is extremely slow, i discovered the bottleneck to be the fact that i am using transactions, after googling for a bit, and playing with the configs, i found out that after autowiring the DefaultMessageListenerContainer and changing the cachelevel to

listenerContainer.setCacheLevelName("CACHE_SESSION") 

my performance increases to about 1500 messages per second while still having transactions.

my problem is now with the producer which is still stuck at about 25 operations per sec, my producer test is simple :

int numOfMessages = getNumberOfMessages();


double startTime = System.currentTimeMillis();

for (int i = 1; i <= numOfMessages; i++) {
    jmsTemplate.convertAndSend("HelloWorld" + i);
}

double endTime = System.currentTimeMillis();

double totalTime=(endTime-startTime)/1000;
System.out.println("Time - "+totalTime+" seconds");
System.out.println("EPS - "+numOfMessages/totalTime);

i'm wondering how to reach similiar performances with the producer, since it now bottlenecks the entire system.

Advantages of HornetQ vs ActiveMQ vs Qpid

I was browsing for an open source messaging software and after some good bit of research I came across these three products. I've taken these out for a preliminary test drive, having had them handle messages for queues and topics, and from what I've read all three of these products are good picks for an Open Source messaging solution for most companies. What I was wondering was what are the advantages that these products may have over one another? What I'm particularly interested in is messaging throughput, including persistent messaging throughput, security, scalability, reliability, support, routing capabilities, administrative options such as metrics and monitoring, and generally just how well each program runs in a large business environment.

What should the Java main method be for a standalone application (for Spring JMS)?

I am interested in creating a Spring standalone application that will run and wait to receive messages from an ActiveMQ queue using Spring JMS. I have searched a lot of places and cannot find a consistent way of implementing the main method for such a standalone application. There appears to be few examples of Spring standalone applications. I have looked at Tomcat, JBoss, ActiveMQ and other examples from the around the web but I have not come to a conclusion so ...

What is the best practice for implementing a main method for a Java application (specifically Spring with JMS) ?

Update: Here's an example from: http://forum.springsource.org/showthread.php?t=48197 Is this the best way of doing this?

public static void main(String args[]) {
        try {
           ApplicationContext context = new ClassPathXmlApplicationContext("spring.xml");
            . . . . .
            Object lock = new Object();
            synchronized (lock) {
                lock.wait();  
            }

        } catch (Exception e) {
            e.printStackTrace();
        }
}

How to use Java JMS with MQseries

I am trying to develop a JMS standalone application to read and write to a Queue on MQSeries. My boss asked me to use pure java JMS (not ibm.mq lib) to do that.

Here is the information that need to make the jms connection:

  mq.hostname=10.10.10.10
  mq.channel=API.CLIENTCHL
  mq.queueManager=MQPETAPI
  mq.port=1422

Do you know how to do that Or do you have any link that teach me to do that.

Active MQ vs JBoss Messaging

I am going to choose a JMS message broker for a project. It is critical that the JMS server is stable and can handle a high load of messages. I have narrowed down the list to include Active MQ and JBoss Messaging.

I was wondering if any of you have any experience with any of these or even better have tried both of them in the same environment. Any link to a research paper or similar would be nice.