java - JMS异常: Queue does not exist

Question

我是 JMS 新手，想要创建一个发送消息的基本 MessageProducer 和异步接收消息的 MessageConsumer。当我运行此代码时，我收到错误消息:

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();
    }

}

创建 MessageConsumer 时，我收到错误消息:

    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)

为什么在创建 MessageConsumer 时出现此错误，但在创建 MessageProducer 时却没有出现此错误。

我使用 ActiveMQServer 作为代理:

服务器.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;
        }
    }
}

Answer 1

我认为，在生产者中，您在设置目的地之前启动连接。之后开始尝试......

// 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);

另一方面，对于消费者，我建议实现 MessageConsumer (而不是 Exception)。实现后，您可以在构造函数中启动消费者

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();

....然后实现onMessage方法

 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();
    }
}