1

I am new in JMS and want to create a basic MessageProducer who sends a message and MessageConsumer who receives the message asynchronously. When I run this code I get error message :

MessageProducer.java

package activemq.test; import java.util.Date; import org.apache.activemq.ActiveMQConnectionFactory; import javax.jms.*; public class MessageProducer{ javax.jms.MessageProducer producer = null; Connection connection = null; Session session = null; public MessageProducer(){ try { // Create a ConnectionFactory ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616"); // Create a Connection connection = connectionFactory.createConnection(); connection.start(); // Create a Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); // Create the destination (Topic or Queue) Destination destination = session.createQueue("TEST.FOO"); // Create a MessageProducer from the Session to the Topic or Queue producer = session.createProducer(destination); producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT); // Create a messages String text = "Hello world! From: MessageProducer"; TextMessage message = session.createTextMessage(text); // Tell the producer to send the message System.out.println("Producer is going to send a message"); producer.send(message); } catch (Exception e) { System.out.println("Caught: " + e); e.printStackTrace(); } } public void sendMessage(){ try { // Create a messages String text = "Hello world! From: " + new Date(); TextMessage message = session.createTextMessage(text); // Tell the producer to send the message System.out.println("Sent message: "+ message.hashCode()); producer.send(message); } catch (Exception e) { e.printStackTrace(); } } public void close(){ // Clean up try { session.close(); connection.close(); } catch (JMSException e) { e.printStackTrace(); } } }

MessageConsumer.java

package activemq.test; import org.apache.activemq.ActiveMQConnectionFactory; import javax.jms.*; public class MessageConsumer implements ExceptionListener{ Connection connection = null; javax.jms.MessageConsumer consumer = null; Session session = null; public MessageConsumer(){ try { // Create a ConnectionFactory ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616"); // Create a Connection connection = connectionFactory.createConnection(); connection.start(); connection.setExceptionListener(this); // Create a Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); // Create the destination (Topic or Queue) Destination destination = session.createQueue("TEST.FOO"); // Create a MessageConsumer from the Session to the Topic or Queue consumer = session.createConsumer(destination); MessageListener listener = new MessageListener() { public void onMessage(Message message) { try { if (message instanceof TextMessage) { TextMessage textMessage = (TextMessage) message; System.out.println("Received message" + textMessage.getText() + "'"); } } catch (JMSException e) { System.out.println("Caught:" + e); e.printStackTrace(); } } }; consumer.setMessageListener(listener); } catch (Exception e) { System.out.println("Caught: " + e); e.printStackTrace(); } } @Override public void onException(JMSException exception) { System.out.println("JMS Exception occured. Shutting down client."); } public void close(){ // Clean up try { consumer.close(); session.close(); connection.close(); } catch (JMSException e) { e.printStackTrace(); } } }

AppMain.java

public class AppMain { public static void main(final String arg[]) throws Exception { MessageProducer msProducer = new MessageProducer(); msProducer.sendMessage(); msProducer.close(); MessageConsumer msConsumer = new MessageConsumer(); msConsumer.close(); } }

When MessageConsumer is created, I get error message:

 Caught: javax.jms.JMSException: AMQ119017: Queue jms.queue.TEST.FOO does not exist javax.jms.JMSException: AMQ119017: Queue jms.queue.TEST.FOO does not exist at org.apache.activemq.util.JMSExceptionSupport.create(JMSExceptionSupport.java:54) at org.apache.activemq.ActiveMQConnection.syncSendPacket(ActiveMQConnection.java:1405) at org.apache.activemq.ActiveMQSession.syncSendPacket(ActiveMQSession.java:1925) at org.apache.activemq.ActiveMQMessageConsumer.<init>(ActiveMQMessageConsumer.java:275) at org.apache.activemq.ActiveMQSession.createConsumer(ActiveMQSession.java:1157) at org.apache.activemq.ActiveMQSession.createConsumer(ActiveMQSession.java:1101) at org.apache.activemq.ActiveMQSession.createConsumer(ActiveMQSession.java:1014) at org.apache.activemq.ActiveMQSession.createConsumer(ActiveMQSession.java:987) at activemq.test.MessageConsumer.<init>(MessageConsumer.java:36) at activemq.test.AppMain.main(AppMain.java:17) Caused by: ActiveMQNonExistentQueueException[errorType=QUEUE_DOES_NOT_EXIST message=AMQ119017: Queue jms.queue.TEST.FOO does not exist] at org.apache.activemq.core.server.impl.ServerSessionImpl.createConsumer(ServerSessionImpl.java:448) at org.apache.activemq.core.protocol.openwire.amq.AMQServerSession.createConsumer(AMQServerSession.java:326) at org.apache.activemq.core.protocol.openwire.amq.AMQConsumer.init(AMQConsumer.java:138) at org.apache.activemq.core.protocol.openwire.amq.AMQSession.createConsumer(AMQSession.java:144) at org.apache.activemq.core.protocol.openwire.OpenWireProtocolManager.addConsumer(OpenWireProtocolManager.java:544) at org.apache.activemq.core.protocol.openwire.OpenWireConnection.processAddConsumer(OpenWireConnection.java:1118) at org.apache.activemq.command.ConsumerInfo.visit(ConsumerInfo.java:347) at org.apache.activemq.core.protocol.openwire.OpenWireConnection.bufferReceived(OpenWireConnection.java:272) at org.apache.activemq.core.remoting.server.impl.RemotingServiceImpl$DelegatingBufferHandler.bufferReceived(RemotingServiceImpl.java:678) at org.apache.activemq.core.remoting.impl.netty.ActiveMQChannelHandler.channelRead(ActiveMQChannelHandler.java:77) at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:332) at io.netty.channel.AbstractChannelHandlerContext.fireChannelRead(AbstractChannelHandlerContext.java:318) at io.netty.channel.DefaultChannelPipeline.fireChannelRead(DefaultChannelPipeline.java:787) at io.netty.channel.nio.AbstractNioByteChannel$NioByteUnsafe.read(AbstractNioByteChannel.java:125) at io.netty.channel.nio.NioEventLoop.processSelectedKey(NioEventLoop.java:507) at io.netty.channel.nio.NioEventLoop.processSelectedKeysOptimized(NioEventLoop.java:464) at io.netty.channel.nio.NioEventLoop.processSelectedKeys(NioEventLoop.java:378) at io.netty.channel.nio.NioEventLoop.run(NioEventLoop.java:350) at io.netty.util.concurrent.SingleThreadEventExecutor$2.run(SingleThreadEventExecutor.java:116) at java.lang.Thread.run(Thread.java:745)

Why I get this error when MessageConsumer is created, but don't get this error when MessageProducer is created.

I use ActiveMQServer as a broker:

Server.java

package activemq.test; import java.util.HashMap; import java.util.HashSet; import java.util.Map; import org.apache.activemq.api.core.TransportConfiguration; import org.apache.activemq.core.config.Configuration; import org.apache.activemq.core.config.impl.ConfigurationImpl; import org.apache.activemq.core.remoting.impl.netty.NettyAcceptorFactory; import org.apache.activemq.core.server.ActiveMQServer; import org.apache.activemq.core.server.ActiveMQServers; public class Server { public static void main(final String arg[]) throws Exception { try { // Step 1. Create the Configuration, and set the properties accordingly Configuration configuration = new ConfigurationImpl(); //we only need this for the server lock file configuration.setJournalDirectory("target/data/journal"); configuration.setPersistenceEnabled(false); // http://activemq.apache.org/what-is-the-difference-between-persistent-and-non-persistent-delivery.html configuration.setSecurityEnabled(false); // http://activemq.apache.org/security.html /** * this map with configuration values is not necessary (it configures the default values). * If you want to modify it to run the example in two different hosts, remember to also * modify the client's Connector at {@link EmbeddedRemoteExample}. */ Map<String, Object> map = new HashMap<String, Object>(); map.put("host", "localhost"); map.put("port", 61616); // https://access.redhat.com/documentation/en-US/JBoss_Enterprise_Application_Platform/5/html/HornetQ_User_Guide/ch14s04.html TransportConfiguration transpConf = new TransportConfiguration(NettyAcceptorFactory.class.getName(),map); HashSet<TransportConfiguration> setTransp = new HashSet<TransportConfiguration>(); setTransp.add(transpConf); configuration.setAcceptorConfigurations(setTransp); // https://github.com/apache/activemq-6/blob/master/activemq-server/src/main/java/org/apache/activemq/spi/core/remoting/Acceptor.java // Step 2. Create and start the server ActiveMQServer server = ActiveMQServers.newActiveMQServer(configuration); server.start(); } catch (Exception e) { e.printStackTrace(); throw e; } } }
Improve this question
14
  • Believe the JVM: You didn't set up the queue. Writing the code isn't sufficient. Commented Apr 29, 2015 at 11:31
  • @duffymo Sorry, but I don't understand. How Should I set the queue? Why don't I get this error when I create MessageProducer? Commented Apr 29, 2015 at 11:35
  • 1
    @Matt You don't get the error on MessageProducer because the message producer does not use the queue at that moment. The consumer does by setting a listener on it, so it tries to listen. Producer just has a producer waiting to be used. If you put the cl.createConsumer() in comment I'm pretty sure the sendMessage will give you that exception. Commented Apr 29, 2015 at 11:42
  • @Juru thanks for response, I get this exception when consumer is created (before the listener is set). Could you give me a hint how can i fix it? Commented Apr 29, 2015 at 11:50
  • 1
    Two doubtful things I have in mind: Something to do with the session and connection not being closed in between creating and using the queue. It doesn't make sense to have the producer and consumer use the same session anyway. Or that the queue is not really created until used (by putting a message on it). Could you do the sendMessage before the create consumer? Commented Apr 29, 2015 at 12:49

1 Answer 1

Reset to default
1
+50

I think, in the producer, you are starting the connection before setting the destination. Try it starting afterwards....

// Create a ConnectionFactory ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory("tcp://localhost:61616"); // Create a Connection connection = connectionFactory.createConnection(); // Create a Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); // Create the destination (Topic or Queue) Destination destination = session.createQueue("TEST.FOO"); // Create a MessageProducer from the Session to the Topic or Queue producer = session.createProducer(destination); producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT); connection.start(); // Create a messages String text = "Hello world! From: MessageProducer"; TextMessage message = session.createTextMessage(text); // Tell the producer to send the message System.out.println("Producer is going to send a message"); producer.send(message);

On the other hand, for the consumer, I suggest to implement MessageConsumer (instead of the Exception). Once implemented, in the constructor you can initiate the consumer 

ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(url); connection = factory.createConnection(); session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE); // Create the destination (Topic or Queue) Destination destination = session.createQueue("TEST.FOO"); // Create a MessageConsumer from the Session to the Topic or Queue consumer = session.createConsumer(destination).setMessageListener(this); connection.start();

.... and then implement the onMessage method

 public void onMessage(Message message) { try { if (message instanceof TextMessage) { TextMessage textMessage = (TextMessage) message; System.out.println("Received message" + textMessage.getText() + "'"); } } catch (JMSException e) { System.out.println("Caught:" + e); e.printStackTrace(); } }
Improve this answer
Sign up to request clarification or add additional context in comments.

2 Comments

I tried it, but it didn't help. I get the same exception.
Have you tried instatiate both producer and consumer, and then send the message?

Start asking to get answers

Find the answer to your question by asking.

Ask question

Explore related questions

See similar questions with these tags.